القائمة
Implement PostgreSQL repository with bulk upsert, soft delete filtering, and fastapi-pagination/filter integration
يُحدَّد Prompt الذي يراجع المصدر أولًا بشكل افتراضي. يمكنك التبديل إلى أمر مباشر أو تنزيل نسخة محلية.
اقرأ SKILL.md وأي ملفات مرافقة يعرضها SkillsMP قبل أن تقرر التثبيت.
التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/agusmdev/fullstack-ai-template --skill fastapi-postgres-repositoryيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
Codebase health scanner and technical debt tracker. Use when the user asks about code quality, technical debt, dead code, large files, god classes, duplicate functions, code smells, naming issues, import cycles, or coupling problems. Also use when asked for a health score, what to fix next, or to create a cleanup plan. Supports 29 languages.
Configure Alembic for async SQLAlchemy migrations with PostgreSQL
Create FastAPI application factory with lifespan, middleware, pagination, and router configuration
عرض SKILL.md
استنادا إلى تصنيف SOC المهني
| name | fastapi-postgres-repository |
| description | Implement PostgreSQL repository with bulk upsert, soft delete filtering, and fastapi-pagination/filter integration |
This skill covers the concrete PostgreSQL repository implementation with all CRUD operations, bulk upserts using ON CONFLICT, soft delete filtering, and integration with fastapi-pagination and fastapi-filter.
Create src/app/common/postgres_repository.py:
from collections.abc import Sequence
from datetime import UTC, datetime
from typing import Any, Generic
from uuid import UUID
from fastapi_filter.contrib.sqlalchemy import Filter
from fastapi_pagination import Params
from fastapi_pagination.bases import AbstractPage
from fastapi_pagination.ext.sqlalchemy import paginate
from pydantic import BaseModel
from sqlalchemy import delete, func, select, update
from sqlalchemy.dialects.postgresql import insert
sqlalchemy.ext.asyncio AsyncSession
app.core.models Base, SoftDeleteMixin
app.core.repository (
AbstractRepository,
CreateSchemaType,
ModelType,
UpdateSchemaType,
)
(
AbstractRepository[ModelType, CreateSchemaType, UpdateSchemaType],
[ModelType, CreateSchemaType, UpdateSchemaType],
):
():
._session = session
._model = model
() -> :
(._model, SoftDeleteMixin)
():
query = select(._model)
._has_soft_delete include_deleted:
query = query.where(._model.deleted_at.is_())
query
() -> ModelType:
data = obj_in.model_dump()
instance = ._model(**data)
._session.add(instance)
._session.commit()
._session.refresh(instance)
instance
() -> ModelType | :
query = ._base_query().where(._model. == )
result = ._session.execute(query)
result.scalar_one_or_none()
() -> [ModelType]:
query = ._base_query()
result = ._session.execute(query)
result.scalars().()
() -> ModelType | :
instance = .get_by_id()
instance:
exclude_unset:
data = obj_in.model_dump(exclude_unset=)
:
data = obj_in.model_dump()
field, value data.items():
(instance, field, value)
._session.commit()
._session.refresh(instance)
instance
() -> :
query = delete(._model).where(._model. == )
result = ._session.execute(query)
._session.commit()
result.rowcount >
() -> AbstractPage[ModelType]:
query = ._base_query()
filter_spec :
query = filter_spec.(query)
query = filter_spec.sort(query)
paginate(._session, query, params)
() -> [ModelType]:
query = ._base_query()
query = filter_spec.(query)
query = filter_spec.sort(query)
result = ._session.execute(query)
result.scalars().()
() -> :
query = select(func.count()).select_from(._model)
._has_soft_delete:
query = query.where(._model.deleted_at.is_())
filter_spec :
subquery = ._base_query()
subquery = filter_spec.(subquery)
query = select(func.count()).select_from(subquery.subquery())
result = ._session.execute(query)
result.scalar_one()
() -> [ModelType]:
objs_in:
[]
data = [obj.model_dump() obj objs_in]
stmt = insert(._model).values(data).returning(._model)
result = ._session.execute(stmt)
._session.commit()
result.scalars().()
() -> [ModelType]:
objs_in:
[]
data = [obj.model_dump() obj objs_in]
stmt = insert(._model).values(data)
update_fields :
update_fields = [
key key data[].keys() key index_elements
]
update_dict = {field: (stmt.excluded, field) field update_fields}
(._model, ):
update_dict[] = func.now()
stmt = stmt.on_conflict_do_update(
index_elements=index_elements,
set_=update_dict,
).returning(._model)
result = ._session.execute(stmt)
._session.commit()
result.scalars().()
() -> :
ids:
query = delete(._model).where(._model..in_(ids))
result = ._session.execute(query)
._session.commit()
result.rowcount
() -> :
._has_soft_delete:
NotImplementedError(
)
query = (
update(._model)
.where(._model. == )
.where(._model.deleted_at.is_())
.values(deleted_at=datetime.now(UTC))
)
result = ._session.execute(query)
._session.commit()
result.rowcount >
() -> :
._has_soft_delete:
NotImplementedError(
)
query = (
update(._model)
.where(._model. == )
.where(._model.deleted_at.is_not())
.values(deleted_at=)
)
result = ._session.execute(query)
._session.commit()
result.rowcount >
() -> ModelType | :
query = ._base_query(include_deleted=).where(._model. == )
result = ._session.execute(query)
result.scalar_one_or_none()
() -> [ModelType]:
query = ._base_query(include_deleted=)
result = ._session.execute(query)
result.scalars().()
() -> :
query = (
select(func.count())
.select_from(._model)
.where(._model. == )
)
._has_soft_delete:
query = query.where(._model.deleted_at.is_())
result = ._session.execute(query)
result.scalar_one() >
() -> [ModelType]:
ids:
[]
query = ._base_query().where(._model..in_(ids))
result = ._session.execute(query)
result.scalars().()
() -> ModelType | :
column = (._model, field)
query = ._base_query().where(column == value)
result = ._session.execute(query)
result.scalar_one_or_none()
() -> [ModelType]:
column = (._model, field)
query = ._base_query().where(column == value)
result = ._session.execute(query)
result.scalars().()
# src/app/items/repository.py
from sqlalchemy.ext.asyncio import AsyncSession
from app.common.postgres_repository import PostgresRepository
from app.items.models import Item
from app.items.schemas import ItemCreate, ItemUpdate
class ItemRepository(PostgresRepository[Item, ItemCreate, ItemUpdate]):
"""Repository for Item entity."""
def __init__(self, session: AsyncSession):
super().__init__(session, Item)
# Add entity-specific methods here if needed
async def get_by_name(self, name: str) -> Item | None:
"""Get item by name."""
return await self.get_by_field("name", name)
async def get_active_items(self) -> list[Item]:
"""Get all active (non-deleted) items."""
return list(await self.get_all())
users = [
UserCreate(email="user1@example.com", name="User 1"),
UserCreate(email="user2@example.com", name="User 2"),
]
# If email exists, update name
await repo.bulk_upsert(
users,
index_elements=["email"],
update_fields=["name"],
)
# Unique constraint on (user_id, product_id)
await repo.bulk_upsert(
cart_items,
index_elements=["user_id", "product_id"],
update_fields=["quantity"],
)
For insert-only (skip existing), use bulk_create with error handling or create a custom method:
async def bulk_create_ignore_conflicts(
self,
objs_in: Sequence[CreateSchemaType],
) -> Sequence[ModelType]:
"""Create records, ignoring conflicts."""
if not objs_in:
return []
data = [obj.model_dump() for obj in objs_in]
stmt = (
insert(self._model)
.values(data)
.on_conflict_do_nothing()
.returning(self._model)
)
result = await self._session.execute(stmt)
await self._session.commit()
return result.scalars().all()
For very large datasets, batch the operations:
async def bulk_create_batched(
self,
objs_in: Sequence[CreateSchemaType],
batch_size: int = 1000,
) -> int:
"""Create records in batches."""
total = 0
for i in range(0, len(objs_in), batch_size):
batch = objs_in[i : i + batch_size]
await self.bulk_create(batch)
total += len(batch)
return total
Ensure index_elements columns have a unique index/constraint:
-- In migration
CREATE UNIQUE INDEX ix_users_email ON users (email);
-- Or
ALTER TABLE users ADD CONSTRAINT uq_users_email UNIQUE (email);
The soft delete filtering happens automatically in _base_query():
# These automatically exclude soft-deleted:
await repo.get_by_id(id)
await repo.get_all()
await repo.get_paginated(params)
# These include soft-deleted:
await repo.get_by_id_with_deleted(id)
await repo.get_all_with_deleted()
fastapi-filter integration works automatically:
# In router
@router.get("")
async def list_items(
filter_spec: ItemFilter = FilterDepends(ItemFilter),
service: ItemService = Depends(get_item_service),
):
return await service.get_paginated(params, filter_spec)
# Filter flows: Router -> Service -> Repository
# Repository calls: filter_spec.filter(query) and filter_spec.sort(query)