Put the logic on the model and the duplication disappears: the ORM domain model

To avoid copy-pasting the same update rule into three places, gather it into a model method

In Part 1 I said the company code was copy-paste all the way down. The spot where that duplication piles up the thickest is right here: where you decide to put your logic.

Treat the model as just a table, and the logic scatters

When I first learned ORM models, I saw them as nothing more than “a table written out as a Python class.” List the columns, wire up the relationships, done. The shape of the data lived there, and everything you actually do with that data went into the service.

Go down that road and the same rule starts repeating in several places. Say you need to update some record: change only the allowed fields, and record “who changed it and when.” You write that once in the update API, again in the job that pulls in external data and refreshes it, and again in the batch that reconciles consistency. All three do the same work, but the code is copy-pasted into three spots. Change the rule in one place and you have to hunt down the other two by hand — and if you miss one, that’s a bug.

The principle I settled on while putting this study guide together is this: put not just the structure of the data but also the behavior tied to that data (the business logic) into the model class. That’s the modeling that object orientation talks about. Once the model stops being a “table” and becomes “an object with behavior,” the rules that used to scatter collect into one place.

BaseSqlModel: write the shared parts once

I made every table model inherit from the same base class. I called it BaseSqlModel.

class BaseSqlModel(SQLModel):
    id: int | None = Field(default=None, primary_key=True)
    inserted_by: str
    inserted_at: datetime
    updated_by: str
    updated_at: datetime

    def insert(self, session_id: str, now: datetime | None = None) -> None:
        ...

    def update(self, session_id: str, now: datetime | None = None, **kwargs) -> None:
        ...

Two things live in here. One is the fields every table shares — id, plus the audit fields that record “when and by whom it was created and changed” (inserted_by/inserted_at/updated_by/updated_at). The other is the helpers that fill those audit fields, insert and update. Call insert and the creator and timestamp go in automatically; call update and the editor and timestamp go in. The one thing to remember is that what gets filled in is a session ID, not a user ID (we record who did it at the session level).

Bake this into the base and you never again have to rewrite the same five fields for every new table, or keep reminding yourself “don’t forget to bump updated_at on edit.” The shared parts are written once, and inheritance carries them along.

Behavior becomes a model method

Once the foundation is laid, you put each model’s own behavior on it as methods. Here’s how it looks for the portfolio model.

class Portfolios(BaseSqlModel, table=True):
    __tablename__ = "portfolio"
    account_id: int = Field(foreign_key="account.id")
    portfolio_name: str = Field(min_length=1, max_length=100, index=True)

    def update_portfolio(self, update_data: dict, session_id: str) -> None:
        for field, value in update_data.items():
            setattr(self, field, value)
        self.updated_by = session_id
        self.updated_at = datetime.now()

That update rule that used to get “copy-pasted all over” now lives in a single update_portfolio. The method encapsulates the work of changing only the allowed fields and filling the audit fields. The update API, the external-data import, the consistency batch — all of them just call this method. When the rule changes, you fix this one spot. No tracking down copy-pasted clones to fix every one.

In the guide I split these model methods into three kinds by role. There’s the write path that changes state or persisted fields (since the change gets saved, it updates the audit fields via self.update(...)); the factory that encapsulates creation rules (a @classmethod that builds and returns a consistent instance); and the read path that computes values derived at query time (current-price valuation, cost-basis calculation, that sort of thing — since it’s not a saved change, it leaves the audit fields alone). The crux of the distinction is “does it fill the audit fields.” If it’s a change that will be persisted, it fills them; if it’s a plain calculation, it doesn’t.

I locked in one more rule alongside this. These model methods stay pure domain. They don’t take things like a session, redis, a repository, or an external API’s response object as arguments. External values come in only as numbers or as other model instances. Set it up this way and you can pull the model out on its own and unit-test it. No DB to spin up — just construct one object, call the method, and check the result.

The service calls the model; it doesn’t copy it

So what does the service do, then? The service does orchestration. It reads data through the repository, draws transaction boundaries, calls external APIs, runs aggregations that sum across several models, and validates consistency. And for calculation, state transitions, and creation on a single model, it doesn’t write that itself — it calls the model method.

This structure really pays off when a router and an async job (a worker) have to do the same thing. The guide’s example is updating an account balance. The balance-sync logic lives in a single service function, and both the balance-update API and the trading worker call that same service.

# router
new_balance = await refresh_account_balance(
    session=session, redis=redis, account=account,
    encrypted_payload=body.encrypted_payload,
    session_id=session_user.session_id,
)

# the worker (task) calls the same service
await sync_account_balance(
    session=self.session, redis=kis_token_redis, account=account,
    encrypted_payload=envelope, session_id="celery-worker", ...
)

The router builds a response, the worker runs in the background — the jobs differ, but the business rule lives in one place. If you’d written the logic inside the router and written the same thing in the worker too, the moment those two drift subtly apart you’ve got a bug.

To sum up, responsibility splits into two layers. The model holds “the rules about itself,” and the service holds “the flow that ties several models together and connects to the outside.” The fact that the same rule never gets written twice is a result of that division of labor. Lift up to the model whatever can be lifted up to the model, and let the service just call it.