メニュー
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コマンドは1行のまま表示されます。コピー前に横へスクロールして全体を確認してください。
ローカルで確認しますか？SkillsMP が現在取得できるファイルをダウンロードできます。
SOC 職業分類に基づく
SKILL.md を表示中
| 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)
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