120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
|
|
from app.core.deps import require_org, require_client_roles, require_write_access, Principal
|
|
from app.models import Brand, Product,AuditLog
|
|
from app.models.enums import ClientRole
|
|
from app.schemas.common import BrandCreate
|
|
|
|
router = APIRouter(prefix="/brands", tags=["brands"])
|
|
|
|
|
|
def _out(b: Brand) -> dict:
|
|
return {"id": str(b.id), "name": b.name, "logo_url": b.logo_url,
|
|
"description": b.description, "product_count": b.product_count,
|
|
"created_at": b.created_at}
|
|
|
|
from fastapi import APIRouter, UploadFile, File
|
|
import os
|
|
import uuid
|
|
|
|
|
|
import os
|
|
|
|
BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
|
|
|
UPLOAD_DIR = os.path.join(BASE_DIR, "_uploads", "brands")
|
|
|
|
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
|
|
|
|
|
@router.post("/upload-logo")
|
|
async def upload_logo(file: UploadFile = File(...)):
|
|
ext = file.filename.split(".")[-1]
|
|
filename = f"{uuid.uuid4()}.{ext}"
|
|
|
|
filepath = os.path.join(UPLOAD_DIR, filename)
|
|
|
|
with open(filepath, "wb") as buffer:
|
|
buffer.write(await file.read())
|
|
|
|
return {
|
|
"url": f"http://localhost:8000/uploads/brands/{filename}"
|
|
}
|
|
|
|
@router.get("")
|
|
async def list_brands(p: Principal = Depends(require_org)):
|
|
brands = await Brand.find(Brand.org_id == str(p.org.id)).to_list()
|
|
return [_out(b) for b in brands]
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_brand(body: BrandCreate,
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
require_write_access(p)
|
|
brand = Brand(org_id=str(p.org.id), name=body.name,
|
|
description=body.description, logo_url=body.logo_url)
|
|
await brand.insert()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="brand.create",
|
|
module="brands",
|
|
details=f"Created brand {brand.name}",
|
|
).insert()
|
|
return _out(brand)
|
|
|
|
|
|
@router.get("/{brand_id}")
|
|
async def get_brand(brand_id: str, p: Principal = Depends(require_org)):
|
|
brand = await Brand.get(brand_id)
|
|
if not brand or brand.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Brand not found")
|
|
products = await Product.find(Product.brand_id == brand_id).to_list()
|
|
return {**_out(brand),
|
|
"products": [{"id": str(x.id), "name": x.name, "sku": x.sku,
|
|
"status": x.status,"image_url": x.image_url,
|
|
"qr_count": x.qr_count,
|
|
"batch_count": x.batch_count,} for x in products]}
|
|
|
|
|
|
@router.put("/{brand_id}")
|
|
async def update_brand(brand_id: str, body: dict,
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin, ClientRole.user))):
|
|
brand = await Brand.get(brand_id)
|
|
if not brand or brand.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Brand not found")
|
|
for k in ("name", "logo_url", "description"):
|
|
if k in body:
|
|
setattr(brand, k, body[k])
|
|
await brand.save()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="brand.update",
|
|
module="brands",
|
|
details=f"Updated brand {brand.name}",
|
|
).insert()
|
|
return _out(brand)
|
|
|
|
|
|
@router.delete("/{brand_id}", status_code=204)
|
|
async def delete_brand(brand_id: str,
|
|
p: Principal = Depends(require_client_roles(ClientRole.admin))):
|
|
brand = await Brand.get(brand_id)
|
|
if not brand or brand.org_id != str(p.org.id):
|
|
raise HTTPException(status.HTTP_404_NOT_FOUND, "Brand not found")
|
|
await brand.delete()
|
|
await AuditLog(
|
|
org_id=str(p.org.id),
|
|
user_id=str(p.user.id),
|
|
user_name=p.user.name,
|
|
user_role=p.client_role,
|
|
action="brand.delete",
|
|
module="brands",
|
|
details=f"Deleted brand {brand.name}",
|
|
).insert()
|