I looked over StackOverflow and were able to find snippets, but not a proper solution to the following problem. I have a model hierarchy in which a job has files (file_models), and files (file_models) have file_events forming the history of a given file.
My project hierarchy's relevant part is this:
[project root]
\ aoget
\ model
\ __init__.py
\ file_event.py
\ file_model.py
\ job.py
\tests
\ test_flie_event.py
\ test_file_model.py
\ ...
pytest.ini
The listed __init__.py file contains this:
from sqlalchemy.orm import scoped_session, sessionmaker, DeclarativeBase
DBSession = scoped_session(sessionmaker())
class Base(DeclarativeBase):
pass
from .file_model import FileModel
from .job import Job
from .file_event import FileEvent
I know this is ugly, but this at least made it work as long as there were no explicit imports in both classes (in case of file_model and file_event I could shape the file in a way in which the import FileModel is not there in FileEvent).
My models inherit from this __init__.Base. Interestingly the application starts and seems to work (don't do any actual persistence yet), but when I want to run a trivial test - construct a FileEvent object -, I get this:
sqlalchemy.exc.InvalidRequestError: One or more mappers failed to initialize - can't proceed with initialization of other mappers. Triggering mapper: 'Mapper[FileModel(file_model)]'. Original exception was: When initializing mapper Mapper[FileModel(file_model)], expression 'Job' failed to locate a name ('Job'). If this is a class name, consider adding this relationship() to the <class 'aoget.model.file_model.FileModel'> class after both dependent classes have been defined
Note that FileModel imports Job and Job imports FileModel.
My test.ini:
[pytest]
pythonpath = aoget
testpaths = tests
[coverage.report]
skip_empty = true
I undertand that the problem is two files mutually referencing each other, but that is expected, as they are relational objects.
How can I organize the models or do something about my test settings to be able to use the models in tests?
Usually I do something like this. Putting together this example seems kind of crazy now after finishing it but it works as your project gets larger, I have run.py just as an example. For larger larger projects I would avoid putting logic into the model classes all together because it really starts to break down. Then you can have all the data model classes in a single file and completely avoid this problem.
src/tproj/model/
src/tproj/model/__init__.py
src/tproj/model/meta.py
src/tproj/model/classes.py
...
src/tproj/model/user.py
src/tproj/model/profile.py
...
src/tproj/run.py
tests/conftest.py
tests/test_profile.py
...
You could store a sessionmaker here as well.
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
def get_engine(environ, echo=False):
return create_engine(f"postgresql+psycopg2://{environ['DB_USERNAME']}:{environ['DB_PASSWORD']}@/{environ['DB_NAME']}", echo=echo)
This is kind of crude but is an easy way to avoid circular imports as long as you do not reference classes in the model scope, see user.py for an example.
from .user import User
from .profile import Profile
Here we use sqlalchemy's system to resolve 'Profile' but we use our classes module to access the actual class in the method scope, create_profile, after all the imports are resolved.
from sqlalchemy import Integer, Column
from sqlalchemy.orm import relationship
from .meta import Base
from . import classes
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
profile = relationship('Profile', back_populates='user', uselist=False)
def create_profile(self, db):
if not self.profile:
profile = classes.Profile(user=self)
db.add(profile)
return profile
from sqlalchemy import Integer, Column, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func, select
from .meta import Base
from . import classes
class Profile(Base):
__tablename__ = 'profiles'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'), nullable=False)
user = relationship('User', back_populates='profile', uselist=False)
@staticmethod
def get_profile_counts(db):
users_without_profile = db.scalar(
select(
func.count(classes.User.id)
).outerjoin(
classes.User.profile
).where(Profile.id == None))
users_with_profile = db.scalar(
select(
func.count(classes.User.id)
).join(
classes.User.profile))
return dict(
users_without=users_without_profile,
users_with=users_with_profile)
import os
from pprint import pprint
from sqlalchemy.orm.session import Session
from sqlalchemy.sql import select, func
from .model.meta import get_engine
from .model import classes
from .model.meta import get_engine, Base
def main():
engine = get_engine(os.environ)
Base.metadata.create_all(engine)
with Session(engine) as session:
user1 = classes.User()
session.add(user1)
user2 = classes.User()
session.add(user2)
profile = user2.create_profile(session)
session.commit()
pprint (classes.Profile.get_profile_counts(session))
if __name__ == '__main__':
main()
import os
import pytest
from sqlalchemy.orm.session import Session
@pytest.fixture
def engine():
from tproj.model.meta import get_engine
engine = get_engine(os.environ)
yield engine
@pytest.fixture
def session(engine):
# Force our classes to be loaded.
import tproj.model.classes
from tproj.model.meta import Base
# HACK
with engine.connect() as conn:
Base.metadata.create_all(conn, checkfirst=True)
conn.commit()
# This requires that you don't actually commit in the code under test.
with Session(engine) as session:
yield session
session.rollback()
@pytest.fixture
def classes(engine):
from tproj.model import classes as classes_
return classes_
import pytest
def test_user(session, classes):
user = classes.User()
session.add(user)
profile = user.create_profile(session)
session.flush()
counts = profile.get_profile_counts(session)
assert counts['users_with'] == 1
assert counts['users_without'] == 0