ZCode
253c922280
Their three commits (730d0bd,5a9b543,387edfb) reached production by push and auto-deployed.387edfbexpects a users.username column that the live database does not have, so POST /api/v1/auth/login was returning HTTP 500. Their schema changes never applied either (projects still has only id, name, description, start_date, pm_user_id, created_at), so nothing depends on the reverted code. This commit sets the tree back to exactlya02b3ea— verified with git diff against it — while keeping history linear: the work itself is preserved on 'pm' at387edfb.
95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
"""Brands CRUD. (name, country) is UNIQUE (uq_brands_name_country)."""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.db.engine import get_db
|
|
from app.models.catalog import Brand, BrandMaterial
|
|
from app.schemas.catalog import BrandCreate, BrandUpdate
|
|
from app.utils.errors import ConflictError, NotFoundError
|
|
from app.utils.pagination import paginate, sort_query
|
|
from app.utils.serialization import to_iso
|
|
|
|
router = APIRouter(prefix="/brands", tags=["brands"])
|
|
|
|
|
|
def brand_to_dict(brand: Brand) -> dict:
|
|
return {
|
|
"id": str(brand.id),
|
|
"name": brand.name,
|
|
"country": brand.country,
|
|
"created_at": to_iso(brand.created_at),
|
|
}
|
|
|
|
|
|
@router.get("")
|
|
async def list_brands(
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(20, ge=1, le=100),
|
|
q: str | None = None,
|
|
country: str | None = None,
|
|
sort: str | None = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
stmt = select(Brand)
|
|
if q:
|
|
stmt = stmt.where(Brand.name.ilike(f"%{q}%"))
|
|
if country:
|
|
stmt = stmt.where(Brand.country == country)
|
|
stmt = sort_query(stmt, Brand, sort)
|
|
items, meta = await paginate(stmt, page, page_size, db)
|
|
return {"data": [brand_to_dict(b) for b in items], "meta": meta}
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_brand(payload: BrandCreate, db: AsyncSession = Depends(get_db)):
|
|
rows = (await db.execute(select(Brand).where(Brand.name == payload.name))).scalars().all()
|
|
if any(b.country == payload.country for b in rows):
|
|
raise ConflictError("برندی با این نام و کشور قبلاً ثبت شده است")
|
|
brand = Brand(name=payload.name, country=payload.country)
|
|
db.add(brand)
|
|
await db.flush()
|
|
return {"data": brand_to_dict(brand)}
|
|
|
|
|
|
@router.get("/{brand_id}")
|
|
async def get_brand(brand_id, db: AsyncSession = Depends(get_db)):
|
|
brand = await db.get(Brand, brand_id)
|
|
if brand is None:
|
|
raise NotFoundError("برند یافت نشد")
|
|
return {"data": brand_to_dict(brand)}
|
|
|
|
|
|
@router.patch("/{brand_id}")
|
|
async def update_brand(brand_id, payload: BrandUpdate, db: AsyncSession = Depends(get_db)):
|
|
brand = await db.get(Brand, brand_id)
|
|
if brand is None:
|
|
raise NotFoundError("برند یافت نشد")
|
|
new_name = payload.name if payload.name is not None else brand.name
|
|
new_country = payload.country if payload.country is not None else brand.country
|
|
rows = (await db.execute(select(Brand).where(Brand.name == new_name, Brand.id != brand_id))).scalars().all()
|
|
if any(b.country == new_country for b in rows):
|
|
raise ConflictError("برندی با این نام و کشور قبلاً ثبت شده است")
|
|
if payload.name is not None:
|
|
brand.name = payload.name
|
|
if payload.country is not None:
|
|
brand.country = payload.country
|
|
await db.flush()
|
|
return {"data": brand_to_dict(brand)}
|
|
|
|
|
|
@router.delete("/{brand_id}")
|
|
async def delete_brand(brand_id, db: AsyncSession = Depends(get_db)):
|
|
brand = await db.get(Brand, brand_id)
|
|
if brand is None:
|
|
raise NotFoundError("برند یافت نشد")
|
|
# Recommendation (b) from P2-4: block instead of silently cascading associations.
|
|
linked = await db.scalar(
|
|
select(func.count()).select_from(BrandMaterial).where(BrandMaterial.brand_id == brand_id)
|
|
)
|
|
if linked:
|
|
raise ConflictError(f"این برند با {linked} متریال مرتبط است و قابل حذف نیست")
|
|
await db.delete(brand)
|
|
await db.flush()
|
|
return {"data": {"id": str(brand_id), "deleted": True}}
|