Spaces:
Paused
Paused
| from fastapi import APIRouter, HTTPException, Depends | |
| from sqlalchemy.ext.asyncio import AsyncSession | |
| from sqlalchemy import select | |
| from typing import List | |
| from ..core.dependencies import get_current_superuser | |
| from ..db.database import get_db | |
| from ..db.models import Branch | |
| from ..db.schemas import BranchCreate, BranchInDB | |
| router = APIRouter() | |
| async def create_branch( | |
| branch: BranchCreate, | |
| current_user = Depends(get_current_superuser), | |
| db: AsyncSession = Depends(get_db) | |
| ) -> BranchInDB: | |
| """Create a new branch (superuser only)""" | |
| db_branch = Branch(**branch.dict()) | |
| db.add(db_branch) | |
| await db.commit() | |
| await db.refresh(db_branch) | |
| return db_branch | |
| async def list_branches( | |
| skip: int = 0, | |
| limit: int = 100, | |
| db: AsyncSession = Depends(get_db) | |
| ) -> List[BranchInDB]: | |
| """List all branches""" | |
| query = select(Branch).offset(skip).limit(limit) | |
| result = await db.execute(query) | |
| return result.scalars().all() | |
| async def get_branch( | |
| branch_id: int, | |
| db: AsyncSession = Depends(get_db) | |
| ) -> BranchInDB: | |
| """Get a specific branch""" | |
| stmt = select(Branch).where(Branch.id == branch_id) | |
| result = await db.execute(stmt) | |
| branch = result.scalar_one_or_none() | |
| if not branch: | |
| raise HTTPException(status_code=404, detail="Branch not found") | |
| return branch | |
| async def update_branch( | |
| branch_id: int, | |
| branch_update: BranchCreate, | |
| current_user = Depends(get_current_superuser), | |
| db: AsyncSession = Depends(get_db) | |
| ) -> BranchInDB: | |
| """Update a branch (superuser only)""" | |
| stmt = select(Branch).where(Branch.id == branch_id) | |
| result = await db.execute(stmt) | |
| branch = result.scalar_one_or_none() | |
| if not branch: | |
| raise HTTPException(status_code=404, detail="Branch not found") | |
| # Update branch fields | |
| for field, value in branch_update.dict().items(): | |
| setattr(branch, field, value) | |
| await db.commit() | |
| await db.refresh(branch) | |
| return branch | |
| async def delete_branch( | |
| branch_id: int, | |
| current_user = Depends(get_current_superuser), | |
| db: AsyncSession = Depends(get_db) | |
| ): | |
| """Delete a branch (superuser only)""" | |
| stmt = select(Branch).where(Branch.id == branch_id) | |
| result = await db.execute(stmt) | |
| branch = result.scalar_one_or_none() | |
| if not branch: | |
| raise HTTPException(status_code=404, detail="Branch not found") | |
| await db.delete(branch) | |
| await db.commit() | |
| return {"status": "success", "message": "Branch deleted"} |