درود دوستان. من توی abc class ریپازیتوریم یه متد list دارم به این صورت. یه کلاس query spec هم دارم که یجور query builder هست
#repository
class AdminRepository:
async def list(
self,
*where: ColumnElement[bool],
load: Sequence[R] = (),
order_by: UnaryExpression | None = None,
limit: int | None = None,
offset: int | None = None,
) -> list[T]:
q = select(self.model).where(*where).options(*self._loaders(load))
if order_by is not None:
q = q.order_by(order_by)
if limit is not None:
q = q.limit(limit)
if offset is not None:
q = q.offset(offset)
result = await self.session.execute(q)
return list(result.scalars().all())
class AdminQuerySpec(BaseQuerySpec):
is_sudo: bool | None = None
orderable_fields = {
'username': Admin.username,
'is_sudo': Admin.is_sudo,
}
searchable_fields = (Admin.username, Admin.id)
def to_clauses(self) -> list[ColumnElement[bool]]:
return [
*super().to_clauses(),
*self._build(
self._eq(Admin.is_sudo, self.is_sudo),
),
]
توی سرویس هم اینطوری دارم ازش استفاده میکنم
# service
class AdminParams(BaseModel):
search: str | None = None
order_by: str = 'username'
limit: int = 50
offset: int = 0
is_sudo: bool | None = None
class AdminService:
async def list(self, filter_: AdminParams) -> tuple[list[Admin], int]:
query = AdminQuerySpec.model_validate(filter_)
total = await self._repo.count(*query.to_clauses())
admins = await self._repo.list(
*query.to_clauses(),
order_by=query.to_order_by(),
limit=filter_.limit,
offset=filter_.offset,
)
return admins, total
ایا این کاری که کردم اصلا درست هست؟ من نمیخواستم برای هر احتمال شرطی که توی ریپو هام دارن یه متد جدا درست کنم. برای همین همون متد list رو با یه کلاس فیلتر dynamic اش کردم
مشکلی که هست حس میکنم یه مقدار از جزئیات repo داره توی service اتفاق میوفته. یعنی همین
query.to_clauses()
query.to_order_by()
یا اشکالی نداره؟
درضمن من قبل از این مستقیما از AdminQuerySpec توی سرویس استفاده میکردم و AdminParams نداشتم. که طی صحبتی که با ai داشتم تصمیم گرفتم جداشون کنم. و توی endpoint هامم از همین AdminParams استفاده کنم جای AdminQuerySpec که بیشتر کارش ساخت db query هست
قبلش به این شکل بود
async def list(self, filter_: AdminQuerySpec) -> tuple[list[Admin], int]:
order_by = filter_.to_order_by()
if order_by is None:
order_by = Admin.username.asc()
clauses = filter_.to_clauses()
total = await self._repo.count(*clauses)
admins = await self._repo.list(
*clauses,
order_by=order_by,
limit=filter_.limit,
offset=filter_.offset,
)
return admins, total
Messageدرود دوستان. من توی abc class ریپازیتوریم یه متد list دارم به این صورت. یه کلاس query spec هم دارم که یجور query builder…
4 October 2026Nearby in the feed
Lthis message
ه