Data Access Goes Through Repositories Only: A Generic BaseRepository

Lock your queries inside repositories and there’s only one place left to change

When you’re writing service code, your hands move before you think. Need a list? One line of select(...). Need a single row? One line of session.get(...). Fast, easy. But once that starts getting sprinkled across every domain and every function, at some point you realize you’ve copy-pasted the same shape of query into ten different places. Want to change one sort rule? Now you have to hunt down all ten and fix every one.

So I set a rule. Data is accessed through repositories, and only through repositories. Service and task bodies don’t use raw queries like select, session.execute, session.get, or update. All data access is delegated to repositories, and the service only says what it wants to fetch.

So I Don’t Rewrite the Same CRUD Every Time

The catch is that if you build a repository per domain, you end up rewriting the same CRUD for every domain too. Fetch, filter, sort, paginate — these look almost identical no matter which entity you’re dealing with.

So I rolled them up into a single generic base class. Declare it as BaseRepository[ModelType] and it attaches to any model, while inside it a QueryBuilder dynamically assembles filters, sorts, pagination, and relationship loading. A domain repository just names its model and inherits the whole toolkit.

class HeroRepository(BaseRepository[Hero]):
    model: Type[Hero] = Hero
    # The basic CRUD comes along for free — not a single line written

To fetch, you just pass a dictionary to find_all. Filter keys take the form "column:operator", and if you omit the operator it defaults to eq. The supported operators are eq/gte/lte/gt/lt/like/startswith/endswith/in/notin.

heroes = await hero_repository.find_all(
    filters={
        "is_active": True,            # operator omitted = eq
        "age:gte": 30,                # 30 or older
        "name:like": "man",          # name contains 'man'
        "team.name:eq": "Team Justice",   # To-One relations via dot
    },
    sorts={"name": "asc"},
    load_with=["powers", "team"],     # eager loading to kill N+1
)

When you also need pagination, use query. This one hands back both the sliced items and the total_count, so it has everything you need to build a list API response.

get or find — What to Do When It’s Missing

Same single-row fetch, but I split it into two paths. get(id) raises a NotFoundException when there’s nothing there; find(id) returns None. Use get for the “this should obviously exist, so error out if it doesn’t” case, and find for “might be there, might not.” The caller’s intent shows through without it having to write its own branching.

Existence checks have a trap too. Don’t use find_by to ask “does it exist?” find_by guarantees a single row, so if the result has two or more rows it blows up with MultipleResultsFound. When all you care about is whether any row exists, use get_total(filters={...}) > 0.

Queries the Basics Can’t Handle Go Their Own Way

BaseRepository doesn’t cover everything. Things like IS NOT NULL, aggregates (count/sum), bulk UPDATE, pulling only columns from another table via JOIN, or filtering on To-Many relations simply can’t be expressed. And it’s more dangerous than it sounds: feed an undefined operator into a key and QueryBuilder silently ignores it, handing you the wrong result.

For these, add a dedicated method to that model’s concrete repository and have the service just call it. The base class never gets touched.

async def find_with_ticker(self) -> list[Stocks]:
    """Only rows where ticker is not NULL. IS NOT NULL can't be expressed as a basic filter."""
    stmt = select(Stocks).where(Stocks.ticker.is_not(None))
    result = await self.session.execute(stmt)
    return list(result.scalars().all())

It’s not that you never write raw queries — it’s that when you do, you cage them inside the repository. From the service’s point of view, wherever the data comes from, it’s still one method call.

A Side Note: Circular References in Schemas

When you’re modeling data and schemas, cycles where things reference each other inevitably show up. A portfolio holds a list of its stocks, and each stock points back at the portfolio. I avoid this with a combination of three things. At the top of the schema file, from __future__ import annotations defers type hints into strings. Schemas from other files get imported only inside an if TYPE_CHECKING: block, with the type hints written as strings. Finally, since those strings have to resolve into actual classes at runtime, I gather every Public schema’s model_rebuild(force=True, _types_namespace=...) call into one place — app/domain/modules/__init__.py — and run them all at once. The point is to make the references resolve only after every module has loaded.

The gist, once more. Data access goes through repositories only, shared CRUD lives in a generic base class, and only the queries the basics can’t handle get pushed down into concrete repositories. Keep the service ignorant of queries, and the day you want to change a single sort rule, there’s just one place left to fix.