Generic Repository with SQLAlchemy and Python

• Download source code - 34.9 KB

Introduction

The Repository Pattern, as well as the Unit of Work Pattern, allows creating an abstraction layer between the data access layer and the business logic layer of an application. The purpose of creating this layer is to isolate the data access layer so that the changes we may operate cannot affect the business logic layer directly. Most of the time, generic repository classes are used, to avoid redundant codes.

Here, we will focus more to create a generic repository class for SQLAlchemy using Python and will use it in a FastAPI project.

Generic Repository

Table

• AppBaseModelOrm base class for common property or columns
• TaskQueue Db table model
• GroupQueue Db table model

models.py:

 import datetime from sqlalchemy import Boolean, Column, Integer, String, \ DateTime, PickleType, Enum as EnumType, JSON from sqlalchemy.dialects.postgresql import UUID from db.database import Base # common fields for all entities class AppBaseModelOrm: id = Column(Integer, primary_key=True, index=True, autoincrement=True) is_active = Column(Boolean, default=True) # soft delete created_by = Column(Integer) updated_by = Column(Integer, default=None) created_datetime = Column(DateTime(timezone=True), default=datetime.datetime.utcnow) updated_datetime = Column(DateTime(timezone=True), \ default=None, onupdate=datetime.datetime.utcnow) account_id = Column(Integer) # tables class TaskQueue(AppBaseModelOrm, Base): __tablename__ = "task_queues" name = Column(String, index=True) class GroupQueue(AppBaseModelOrm, Base): __tablename__ = "group_queues" name = Column(String, index=True)

Repository Class

TableRepository is the base repo class which contains some basic operations/methods, like:

• Read data from a table
• Add/update/delete rows from/to a table

At the constructor level, the repository class expects:

• db:Session DB session object
• entity:object Table entity

table_repo.py:

 from sqlalchemy import and_ from sqlalchemy.orm import Session from sqlalchemy.sql.expression import false from datetime import datetime class TableRepository: entity:object = NotImplementedError db:Session = NotImplementedError def __init__(self, db:Session, entity:object): self.db = db self.entity = entity def get_all(self): return self.db.query(self.entity) def get_by_id(self, id:int): return self.db.query(self.entity).filter(self.entity.id==id).one() def find_by_id(self, id:int): return self.db.query(self.entity).filter(self.entity.id==id).first() def get_actives(self): return self.db.query(self.entity).filter(self.entity.is_active==True) def get_by_account_id(self, account_id:int): return self.db.query(self.entity).filter(self.entity.account_id==account_id) def get_actives_by_account_id(self, account_id:int): return self.db.query(self.entity).filter\ (self.entity.is_active==True, self.entity.account_id==account_id) def get_by_create_datetime_range(self, from_datetime:datetime, to_datetime:datetime): data = self.db.query(self.entity).filter\ (self.entity.created_datetime >= from_datetime, \ self.entity.created_datetime <= to_datetime) return data def add(self, entity, created_by_user_id:int = None): entity.created_by = created_by_user_id self.db.add(entity) def update(self, entity, updated_by_user_id:int = None): entity.updated_by = updated_by_user_id def delete(self, entity, deleted_by_user_id:int = None): entity.is_active = False self.update(entity, updated_by_user_id=deleted_by_user_id) def permanent_delete(self, entity): self.db.delete(entity) 

Using Repository Class

Here, we are using TableRepository repo class without any inheritance, for TaskQueue entity and read/write table task_queues.

task_queue.py:

 from typing import Optional, List from fastapi import FastAPI, Request, Depends from fastapi_utils.cbv import cbv from fastapi_utils.inferring_router import InferringRouter from sqlalchemy.orm import Session from app import schemas from app.depends.db_depend import get_db from app.depends.auth_depend import get_current_user, CurrentUser from db import models from db.table_repo import TableRepository router = InferringRouter() @cbv(router) class TaskQueue: db: Session = Depends(get_db) current_user:CurrentUser = Depends(get_current_user) @router.get("/", response_model=List[schemas.TaskQueueSchema]) def get_all(self): repo = TableRepository(self.db, models.TaskQueue) items = repo.get_all().all() return items @router.get("/actives", response_model=List[schemas.TaskQueueSchema]) def get_actives(self): repo = TableRepository(self.db, models.TaskQueue) items = repo.get_actives().all() return items @router.get("/account/{account_id}", response_model=List[schemas.TaskQueueSchema]) def get_by_account(self, account_id: int): repo = TableRepository(self.db, models.TaskQueue) items = repo.get_by_account_id(account_id).all() return items @router.get("/account/{account_id}/actives", \ response_model=List[schemas.TaskQueueSchema]) def get_actives_by_account(self, account_id: int): repo = TableRepository(self.db, models.TaskQueue) items = repo.get_actives_by_account_id(account_id).all() return items @router.get("/{id}", response_model=schemas.TaskQueueSchema) def get_by_id(self, id: int): repo = TableRepository(self.db, models.TaskQueue) item = repo.get_by_id(id) return item @router.get("/find/{id}", response_model=schemas.TaskQueueSchema) def find_by_id(self, id: int): '''can be null''' repo = TableRepository(self.db, models.TaskQueue) item = repo.find_by_id(id) return item @router.post("/", response_model=schemas.TaskQueueSchema) def post_item(self, model: schemas.TaskQueueCreate): item = models.TaskQueue(name=model.name, account_id=model.account_id) repo = TableRepository(self.db, models.TaskQueue) repo.add(item, self.current_user.id) self.db.commit() self.db.refresh(item) return item @router.put("/{id}", response_model=schemas.TaskQueueSchema) def put_item(self, id:int, model: schemas.TaskQueueUpdate): '''can be null''' repo = TableRepository(self.db, models.TaskQueue) item = repo.find_by_id(id) if item: item.name = model.name repo.update(item, self.current_user.id) self.db.commit() self.db.refresh(item) return item @router.delete("/{id}", response_model=schemas.TaskQueueSchema) def delete_item(self, id: int): '''can be null''' repo = TableRepository(self.db, models.TaskQueue) item = repo.find_by_id(id) if item: repo.delete(item, self.current_user.id) self.db.commit() self.db.refresh(item) return item @router.delete("/permanent/{id}", response_model=schemas.TaskQueueSchema) def permanent_delete_item(self, id: int): '''can be null''' repo = TableRepository(self.db, models.TaskQueue) item = repo.find_by_id(id) if item: repo.permanent_delete(item) self.db.commit() return item

Using Repository Class as Base

GroupQueueCrud is inheriting the repo class TableRepository for GroupQueue entity and read/write table group_queues:

Inheriting Repository Class

group_queue_crud.py:

 from sqlalchemy.orm import Session from app import schemas from db import models from db.table_repo import TableRepository class GroupQueueCrud(TableRepository): def __init__(self, db:Session): super().__init__(db=db, entity=models.GroupQueue)

Using the CRUD Class

group_queue.py:

 from datetime import datetime from typing import Optional, List from fastapi import FastAPI, Request, Depends, Query from fastapi_utils.cbv import cbv from fastapi_utils.inferring_router import InferringRouter from sqlalchemy.orm import Session from app import schemas from app.depends.db_depend import get_db from app.depends.auth_depend import get_current_user, CurrentUser from app.cruds.group_queue_crud import GroupQueueCrud from db import models router = InferringRouter() @cbv(router) class GroupQueue: db: Session = Depends(get_db) current_user:CurrentUser = Depends(get_current_user) @router.get("/{id}", response_model=schemas.GroupQueueSchema) def get_by_id(self, id: int): repo = GroupQueueCrud(self.db) item = repo.find_by_id(id) return item @router.post("/", response_model=schemas.GroupQueueSchema) def post_item(self, model: schemas.GroupQueueCreate): item = models.GroupQueue(name=model.name, account_id=model.account_id) repo = GroupQueueCrud(self.db) repo.add(item, self.current_user.id) self.db.commit() self.db.refresh(item) return item @router.put("/{id}", response_model=schemas.GroupQueueSchema) def put_item(self, id:int, model: schemas.GroupQueueUpdate): '''can be null''' repo = GroupQueueCrud(self.db) item = repo.find_by_id(id) if item: item.name = model.name repo.update(item, self.current_user.id) self.db.commit() self.db.refresh(item) return item @router.delete("/{id}", response_model=schemas.GroupQueueSchema) def delete_item(self, id: int): '''can be null''' repo = GroupQueueCrud(self.db) item = repo.find_by_id(id) if item: repo.delete(item, self.current_user.id) self.db.commit() self.db.refresh(item) return item 

Using the Code

 Go to backend folder Open cmd Type docker-compose up -d \backend> docker-compose up -d project will run http://localhost:4003 Go to Api Doc http://localhost:4003/docs#/

Check Task Queue and Group Queue sections:

References

• https://github.com/pseudobabble/repository

History

• 5^th July, 2022: Initial version