45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Generic, TypeVar
|
|
|
|
from sqlalchemy import Select, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
ModelT = TypeVar("ModelT")
|
|
|
|
|
|
class Repository(Generic[ModelT]):
|
|
model: type[ModelT]
|
|
|
|
def __init__(self, session: Session) -> None:
|
|
self.session = session
|
|
|
|
def get(self, item_id: str) -> ModelT | None:
|
|
return self.session.get(self.model, item_id)
|
|
|
|
search_columns: tuple[str, ...] = ()
|
|
|
|
def _search_statement(self, search: str | None = None) -> Select[tuple[ModelT]]:
|
|
statement: Select[tuple[ModelT]] = select(self.model)
|
|
if search and self.search_columns:
|
|
term = f"%{search.strip()}%"
|
|
filters = [getattr(self.model, column).ilike(term) for column in self.search_columns]
|
|
statement = statement.where(or_(*filters))
|
|
return statement
|
|
|
|
def list(self, limit: int = 20, offset: int = 0, search: str | None = None) -> list[ModelT]:
|
|
statement = self._search_statement(search).offset(offset).limit(limit)
|
|
return list(self.session.scalars(statement))
|
|
|
|
def count(self, search: str | None = None) -> int:
|
|
subquery = self._search_statement(search).subquery()
|
|
return self.session.scalar(select(func.count()).select_from(subquery)) or 0
|
|
|
|
def add(self, item: ModelT) -> ModelT:
|
|
self.session.add(item)
|
|
self.session.flush()
|
|
return item
|
|
|
|
def delete(self, item: ModelT) -> None:
|
|
self.session.delete(item)
|
|
self.session.flush()
|