Step 6 - Data Access Layer (BaseRepository)
Implement asynchronous CRUD, keyset pagination, and secure search out of the box with zero boilerplate.
Writing the same SELECT, INSERT, and UPDATE queries for every model is tedious. ZCore's BaseRepository eliminates this boilerplate.
Create the Repository
Open tasks/repositories.py. Inherit from BaseRepository and pass your model type:
# tasks/repositories.py
from sqlalchemy.ext.asyncio import AsyncSession
from zcore import BaseRepository
from .models import Task
class TaskRepo(BaseRepository[Task]):
def __init__(self, db: AsyncSession):
super().__init__(model=Task, db=db)What You Just Unlocked
Out-of-the-box features: By writing this empty class, TaskRepo now has access to:
- CRUD & Mutations:
create,create_multi,get,get_by_ids,exist,count,update, anddelete(with direct primary key or model instance support). - Soft-Delete & Restoration: Built-in
restoreand atomic batchrestore_multifor recovering soft-deleted records, plus aforce=Trueflag ondelete/delete_multifor permanent hard-deletes. - Batch Operations:
update_multi,delete_multi(with dialect-awareRETURNINGoptimizations). - Pagination:
get_list(supports both Page-Number and high-performance Keyset Cursor pagination). - Dynamic Search & Projection:
searchwith selective column loading (fieldsviaload_only), custom SQLAlchemy options (options), depth-limit validation, inverted operators (not_like,not_in,not_between,is_not_null), and booleannotgrouping.
All methods are fully asynchronous and type-hinted.
No Lock-In
If you have a complex query that requires a raw SQL JOIN or a CTE, you can simply write a new custom method or override any existing CRUD method inside TaskRepo. Because ZCore never hides the AsyncSession from you, you have full access to SQLAlchemy's power directly inside your repository.
Next, we will add our business logic layer using BaseService.