import sys from typing import Annotated from fastapi import APIRouter, Query, Depends, HTTPException # from sqlmodel import SQLModel, Field from sqlmodel import select, and_ from pydantic import BaseModel, Field # from sqlalchemy import select import alchemy import utils from dependencies import get_session from utils import update_item, create_item from .status_model import Status from .user import ACTIVE_USER PRIMARY_ANNOTATION = utils.make_primary_annotation('Country') # ---------------------------------------------------------------- # Models # class LocationBase(SQLModel): # code: str = Field(max_length=30) # description: str | None = Field(max_length=256) # notes: str | None # # # class LocationCodeBase(SQLModel): # code: str = Field(max_length=8) # description: str | None= Field(max_length=256) # notes: str | None # # # class CountryBase(SQLModel): # code: str = Field(max_length=2) # name: str = Field(max_length=80) # notes: str | None # class ILocationCreate(LocationBase): # description: str | None = None # notes: str | None = None # # # class ILocationCodeCreate(LocationCode): # description: str | None = None # notes: str | None = None # class LocationCreate(LocationBase): # description: str | None = None # notes: str | None # # # class LocationCodeCreate(LocationCodeBase): # description: str | None = None # notes: str | None = None # locations: list[LocationCreate] | None = None # # # class CountryCreate(CountryBase): # notes: str | None = None # location_codes: list[LocationCodeCreate] | None = None # class Country(CountryBase): # id: int | None = Field(default=None, primary_key=True) # # # class CountryCreate(CountryBase): # notes: str | None = None class Contact(BaseModel): code: str address: str | None notes: str | None class Location(BaseModel): id: int code: str description: str | None notes: str | None status: Status contact: Contact | None class LocationCode(BaseModel): id: int code: str description: str | None notes: str | None status: Status locations: list[Location] contact: Contact | None class Country(BaseModel): id: int code: str name: str notes: str | None status: Status location_codes: list[LocationCode] | None class LocationCreate(BaseModel): code: str = Field(max_length=30) description: str | None = None notes: str | None = None class LocationCodeCreate(BaseModel): code: str = Field(max_length=8) description: str | None = None notes: str | None = None locations: list[LocationCreate] | None = None class CountryCreate(BaseModel): code: str = Field(max_length=2) name: str = Field(max_length=80) notes: str | None = None location_codes: list[LocationCodeCreate] | None = None class CountryUpdate(BaseModel): code: str | None = Field(default=None, max_length=2) name: str | None = Field(default=None, max_length=80) notes: str | None = None # class CountryUpdate(CountryBase): # code: str | None = None # name: str | None = None # notes: str | None = None # ---------------------------------------------------------------- # Routes router = APIRouter(prefix="/location", tags=["location"]) @router.get("/", response_model=list[Country]) async def get_countries( offset: int = 0, limit: Annotated[int, Query] = 100, session=Depends(get_session)): """Get list of all countries""" if limit < 1: limit = sys.maxsize result = session.exec(select(alchemy.Country).offset(offset).limit(limit)).all() return result @router.get("/{country}", response_model=Country, responses={404: {"description": "Not found"}}) async def get_country( country: PRIMARY_ANNOTATION, session=Depends(get_session)): return utils.get_single_record(session, alchemy.Country, country) @router.post("/", response_model=Country) async def create_country( country: CountryCreate, current_user: ACTIVE_USER, session=Depends(get_session)): item = CountryCreate.model_validate(country).model_dump(exclude_unset=True) if session.scalar(select(alchemy.Country).where(alchemy.Country.code == item['code'])): raise HTTPException(status_code=422, detail=[dict(msg=f"Country {item['code']!r} already exists", type="Database Integrity Error")]) country = alchemy.Country(**{k: v for k, v in item.items() if k in ('code', 'name', 'notes')}) country.code = country.code.upper() country._user__id = current_user.id for i in item.get('location_codes', []): location_code = alchemy.LocationCode(**{k: v for k, v in i.items() if k in ('code', 'name', 'notes')}) location_code.code = location_code.code.upper() location_code._user__id = current_user.id for j in i.get('locations', []): location = alchemy.Location(**j) location._user__id = current_user.id location_code.locations.append(location) country.location_codes.append(location_code) session.add(country) try: session.commit() except Exception as exc: raise HTTPException(status_code=422, detail=[dict(msg=', '.join(exc.args), type="Database Error")]) session.refresh(country) return country @router.patch("/{country}", response_model=Country, responses={404: {"description": "Not found"}}) async def update_country( country: PRIMARY_ANNOTATION, data: CountryUpdate, current_user: ACTIVE_USER, session=Depends(get_session)): return update_item( session=session, current_user=current_user, item=utils.get_single_record(session, alchemy.Country, 'Country', country), data=data) @router.put("/{country}/activate", response_model=Country, responses={404: {"description": "Not found"}}) async def activate_country( country: PRIMARY_ANNOTATION, current_user: ACTIVE_USER, session=Depends(get_session)): return utils.set_item_status( session=session, current_user=current_user, item=utils.get_single_record(session, alchemy.Country, 'Country', country), status='A') @router.put("/{country}/deactivate", response_model=Country, responses={404: {"description": "Not found"}}) async def deactivate_country( country: PRIMARY_ANNOTATION, current_user: ACTIVE_USER, session=Depends(get_session)): return utils.set_item_status( session=session, current_user=current_user, item=utils.get_single_record(session, alchemy.Country, 'Country', country), status='I') @router.post("/{country}", response_model=LocationCode,) async def create_location_code( country: PRIMARY_ANNOTATION, data: LocationCodeCreate, current_user: ACTIVE_USER, session=Depends(get_session)): item = LocationCodeCreate.model_validate(data).model_dump(exclude_unset=True) if session.scalar(select(alchemy.LocationCode).where(alchemy.LocationCode.code == data.code)): raise HTTPException(status_code=422, detail=[dict(msg=f"Location Code {data.code!r} already exists", type="Database Integrity Error")]) country = utils.get_single_record(session, alchemy.Country, 'Country', country) location_code = alchemy.LocationCode(**{k: v for k, v in item.items() if k in ('code', 'description', 'notes')}, country=country) location_code._user__id = current_user.id for i in item.get('locations', []): location = alchemy.Location(**i) location._user__id = current_user.id location_code.locations.append(location) session.add(location_code) try: session.commit() except Exception as exc: raise HTTPException(status_code=422, detail=[dict(msg=', '.join(exc.args), type="Database Error")]) session.refresh(location_code) return location_code @router.post("/{country}/{location_code}", response_model=LocationCode,) async def create_location( country: PRIMARY_ANNOTATION, location_code: PRIMARY_ANNOTATION, data: LocationCode, current_user: ACTIVE_USER, session=Depends(get_session)): location_code = utils.get_single_record(session, alchemy.Country, 'Country', country) item = LocationCreate.model_validate(data).model_dump(exclude_unset=True) if session.scalar(select(alchemy.LocationCode).where( and_(alchemy.Location.location_code == location_code, alchemy.Location.code == data.code))): raise HTTPException(status_code=422, detail=[dict(msg=f"Location '{location_code.code}/{data.code}' already exists", type="Database Integrity Error")]) location = alchemy.Location(**{k: v for k, v in item.items() if k in ('code', 'description', 'notes')}, location_code=location_code) location._user__id = current_user.id session.add(location) try: session.commit() except Exception as exc: raise HTTPException(status_code=422, detail=[dict(msg=', '.join(exc.args), type="Database Error")]) session.refresh(location_code) return location_code