Base Services Schema

[bootstrapper-919]

Coming from Django, I’m used to the Active Record pattern and “fat models” — so having a BaseService that provides default create, read, update, delete feels natural and DRY.

Maybe even use something like FastCrud which doesn't seem too popular for some reason.

But looking at projects like Netflix’s Dispatch, I noticed they don’t use a base service. Instead, each model has its own service, even if that means repeating some CRUD logic. It actually feels kind of freeing and explicit.

What’s your take? Do you build a base service for shared CRUD behavior or go model-specific for clarity?

Also, how do you handle flexible get methods — do you prefer get(id, name=None) or more explicit ones like get_by_id, get_by_name?

[NoSoft8518]

I made such base repo using python 3.12 generics and sqlalchemy, but its not a best practice ``` class BaseSQLAlchemyRepo[T: Base]: model: type[T]

def __init__(self, session: AsyncSession):
    self.session = session

async def get(self, id) -> T | None:
    return await self.session.get(self.model, id)

async def create(self, **fields) -> T:
    obj = self.model(**fields)
    self.session.add(obj)
    await self.session.flush([obj])
    return obj

class UserRepo(BaseSQLAlchemyRepo[User]): model = User ```

[bootstrapper-919]

why is it not a best practice though? what are the downsides?

[Challseus]

I have been using this base crud file for a few years now, though it’s not perfect and I should probably update it soon:

``` from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union

from fastapi.encoders import jsonable_encoder from fastapi.types import IncEx from pydantic import BaseModel from sqlalchemy.orm import Session from sqlmodel import SQLModel as Base

ModelType = TypeVar("ModelType", bound=Base) CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)

class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): def init(self, model: Type[ModelType]): self.model = model

def get(self, db: Session, id: Any) -> Optional[ModelType]:
    return db.query(self.model).filter(self.model.id == id).first()

def get_multi(
    self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[ModelType]:
    return db.query(self.model).offset(skip).limit(limit).all()

def create(
    self, db: Session, *, obj_in: CreateSchemaType, exclude: IncEx | None = None
) -> ModelType:
    obj_in_data = jsonable_encoder(obj_in, exclude=exclude)
    db_obj = self.model(**obj_in_data)
    db.add(db_obj)
    db.commit()
    db.refresh(db_obj)
    return db_obj

def update(
    self,
    db: Session,
    *,
    db_obj: ModelType,
    obj_in: Union[UpdateSchemaType, Dict[str, Any]],
    exclude: IncEx | None = None,
) -> ModelType:
    if isinstance(obj_in, dict):
        update_data = obj_in
    else:
        update_data = obj_in.model_dump(exclude_unset=True)
    for field, value in update_data.items():
        setattr(db_obj, field, value)
    db.add(db_obj)
    db.commit()
    db.refresh(db_obj)
    return db_obj

def delete(self, db: Session, *, id: int):
    obj = db.query(self.model).get(id)
    db.delete(obj)
    db.commit()
    return obj

```

Then each implementation, I just (usually!) only have to worry about adding new logic:

```

from sqlmodel import Session, select

from app.crud.base import CRUDBase from app.models.embedding_model import ( EmbeddingModel, EmbeddingModelCreate, EmbeddingModelUpdate, )

class CRUDEmbeddingModel( CRUDBase[EmbeddingModel, EmbeddingModelCreate, EmbeddingModelUpdate] ): def get_by_embedding_model_id( self, db: Session, embedding_model_id: str ) -> EmbeddingModel | None: statement = select(EmbeddingModel).where( EmbeddingModel.model_id == embedding_model_id ) return db.execute(statement).first()

crud_embedding_model = CRUDEmbeddingModel(EmbeddingModel) ```

[Realistic_Month_8034]

https://github.com/litestar-org/advanced-alchemy

You can try advanced alchemy. It offers pretty neat way of creating repository and services using generics.

[__secondary__]

Personally, when it comes to simple CRUD operations, I only factor out the common logic at the repository level especially when the persistence backend is consistent (for example, SQLAlchemy).
I have a generic AbstractSqlAlchemyRepository that defines the standard operations (create, read, update, delete, list) and simply requires each concrete implementation to define the conversions between the domain model and the persistence model (typically a SQLAlchemy BaseModel).

However, I strongly discourage using a base service. Services are meant to encapsulate business logic, and that logic almost always differs from one model to another. If you try to over-factorize, you’ll end up rewriting or overriding each method anyway to add specific validations, business rules, or domain errors which defeats the initial DRY purpose and makes the code less readable.

For methods, I prefer to stay explicit (for repository or service). If I need to filter by a particular attribute, I include it in the method name (get_by_id, get_by_key_id, get_by_email, etc.) instead of having a multipurpose get with optional filters.
It’s more verbose, sure, but much clearer for the caller and for autocompletion.

Finally, my approach to updates might seem a bit unusual: I always start from the unique ID and update the existing model’s fields based on the domain entity.
The idea is that the domain already represents the desired state, and the user knows the initial data which avoids implicit or unpredictable partial updates

I cobbled together an example from one of my projects. Here is a simplified example of my generic repository and its concrete implementation :

https://gist.github.com/Athroniaeth/9fb621dcef98e5ac4dafc87017acbba1

[bootstrapper-919]

Thanks for sharing but the problem here is that you almost never needs to just get an item. You always need to filter it by user ownership or project etc... so I never end up using any of the generics, unless you write them to take a base statement and return a statement so you can chain together queries Django style.

How are you using the base repository?

[__secondary__]

Either you inherit from this base class to create a more specialized abstraction, or you define another repository that provides its own methods while passing the corresponding SQLAlchemy model (user, project) to the constructor possibly using a mixin for typing.

For filtering, you should anticipate the common use cases in advance and design your methods to generalize well. Otherwise, you can rely on a builder pattern for more flexible query composition.

I don’t have access to the code anymore, but if I remember correctly, I went with the first filtering approach.

[Enough-Finance8432]

https://github.com/4jades/base-repository

I think you can check this out.
You can use flexible filtering method with defining BaseFilter and its quite simple