Files
api/src/database/query_builder.py
T

68 lines
2.2 KiB
Python

from typing import Optional, Type
from sqlalchemy import select, or_, and_
from database import models
class QueryBuilder:
def __init__(self, model: Type[models.BaseModel]):
self.model = model
def get_simple_query(
self,
extra_select: list | None = None,
created_by_id: int | None = None,
order_by: list | None = None,
only_public: bool | None = False,
only_my: bool | None = False,
include_others_public: bool | None = False,
url_slug: str | None = None,
include_deleted: bool = False
):
if not extra_select:
extra_select = []
query = select(self.model, *extra_select)
if not include_deleted and hasattr(self.model, "deleted"):
query = query.filter(self.model.deleted.is_(False))
ownership_filters = []
my_filters = []
if only_public and hasattr(self.model, "is_public"):
my_filters.append(self.model.is_public.is_(True))
if hasattr(self.model, "url_slug"):
my_filters.append(self.model.url_slug != '')
if only_my:
if created_by_id and hasattr(self.model, "created_by_id"):
my_filters.append(self.model.created_by_id == created_by_id)
others_filters = []
if include_others_public:
if hasattr(self.model, "is_public"):
others_filters.append(self.model.is_public.is_(True))
if hasattr(self.model, "created_by_id"):
others_filters.append(or_(
self.model.created_by_id != created_by_id,
self.model.created_by_id.is_(None)
))
if my_filters:
ownership_filters.append(and_(*my_filters))
if others_filters:
ownership_filters.append(and_(*others_filters))
query = query.filter(or_(*ownership_filters))
if url_slug is not None and hasattr(self.model, 'url_slug'):
query = query.filter(self.model.url_slug == url_slug)
if order_by:
query = query.order_by(*order_by)
elif hasattr(self.model, "name"):
query = query.order_by(self.model.name)
return query