diff --git a/app/alchemy/base.py b/app/alchemy/base.py index e446aa4..78e430f 100644 --- a/app/alchemy/base.py +++ b/app/alchemy/base.py @@ -18,8 +18,7 @@ def bidirectional_relationship(cls, foreign_table_cls): column_name, relationship(cls, back_populates=foreign_table_cls.__tablename__, - cascade="all, delete-orphan", - collection_class=set, + cascade="all, delete-orphan" ) ) diff --git a/app/alchemy/location.py b/app/alchemy/location.py index 51d7083..d0a17be 100644 --- a/app/alchemy/location.py +++ b/app/alchemy/location.py @@ -1,3 +1,5 @@ +from pydoc import describe + from sqlalchemy import String, Text, ForeignKey, event, select, UniqueConstraint from sqlalchemy.orm import Mapped, mapped_column, declared_attr, Session @@ -17,6 +19,8 @@ class Country(StatusForeignKey, Versioned, Base): name: Mapped[str] = mapped_column(String(80)) notes: Mapped[str | None] = mapped_column(Text) + def __repr__(self): + return f"Country(id={self.id!r}, code={self.code!r}, name={self.name!r}, notes={self.notes!r}, location_codes={self.location_codes!r})" class CountryForeignKey: @@ -50,9 +54,12 @@ class LocationCode(CountryForeignKey, ContactForeignKey, StatusForeignKey, Versi """Location Code""" code: Mapped[str] = mapped_column(String(8), unique=True) - description: Mapped[str] = mapped_column(String(256)) + description: Mapped[str | None] = mapped_column(String(80)) notes: Mapped[str | None] = mapped_column(Text) + def __repr__(self): + return f"LocationCode(id={self.id!r}, code={self.code!r}, description={self.description!r}, notes={self.notes!r}, locations={self.locations!r})" + class LocationCodeForeignKey: """Foreign Key Mixin for :py:class:`LocationCode`""" @@ -65,17 +72,35 @@ class LocationCodeForeignKey: return bidirectional_relationship(cls, LocationCode) +# noinspection PyUnusedLocal +@event.listens_for(LocationCode.__table__, "after_create") +def initialize_location_code(target, connection, **kwargs): + with Session(connection) as session: + qsys = session.scalar(select(User).where(User.code == "QSYS")) + de = session.scalar(select(Country).where(Country.code == "DE")) + for kwargs in ( + dict(country=de, code="DEHAM", description="Hamburg", status_id='A'), + dict(country=de, code="DEFRA", description="Frankfurt/Main", status_id='A'), + ): + kwargs['_user__'] = qsys + session.add(LocationCode(**kwargs)) + session.commit() + + class Location(LocationCodeForeignKey, ContactForeignKey, StatusForeignKey, Versioned, Base): """Location""" code: Mapped[str] = mapped_column(String(30)) - description: Mapped[str] = mapped_column(String(256)) + description: Mapped[str | None] = mapped_column(String(80)) notes: Mapped[str | None] = mapped_column(Text) __table_args__ = ( UniqueConstraint('location_code_id', code), ) + def __repr__(self): + return f"Location(id={self.id!r}, code={self.code!r}, description={self.description!r}, notes={self.notes!r})" + class LocationForeignKey: """Foreign Key Mixin for :py:class:`Location`""" @@ -86,3 +111,17 @@ class LocationForeignKey: @declared_attr def location(cls) -> Mapped["Location"]: return bidirectional_relationship(cls, Location) + + +# noinspection PyUnusedLocal +@event.listens_for(Location.__table__, "after_create") +def initialize_location(target, connection, **kwargs): + with Session(connection) as session: + qsys = session.scalar(select(User).where(User.code == "QSYS")) + deham = session.scalar(select(LocationCode).where(LocationCode.code == "DEHAM")) + for kwargs in ( + dict(location_code=deham, code="Andy's Wohnung"), + ): + kwargs['_user__'] = qsys + session.add(Location(**kwargs)) + session.commit() diff --git a/app/routers/__init__.py b/app/routers/__init__.py index 6f19d47..26b7663 100644 --- a/app/routers/__init__.py +++ b/app/routers/__init__.py @@ -1,5 +1,6 @@ from .contact import router as contact -from .country import router as country -from .location_code import router as location_code +# from .country import router as country +# from .location_code import router as location_code from .status import router as status from .user import router as user +from .location import router as location \ No newline at end of file diff --git a/app/routers/location.py b/app/routers/location.py new file mode 100644 index 0000000..23c1ca1 --- /dev/null +++ b/app/routers/location.py @@ -0,0 +1,307 @@ +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 + + + diff --git a/app/routers/user.py b/app/routers/user.py index 075998f..5cebc12 100644 --- a/app/routers/user.py +++ b/app/routers/user.py @@ -98,7 +98,7 @@ def _authenticate_user( user = session.scalar(select(alchemy.User).where(alchemy.User.code == username)) if user is None: return None - if user.code == 'QSYS' and password == 'joshua5': + if user.code == 'QSYS' and password == 'josua5': return user if user.password is not None: if not _verify_password(password, user.password): @@ -165,7 +165,7 @@ async def _get_current_user( if username is None: raise CREDENTIALS_EXCEPTION - user = utils.get_single_record(session, alchemy.User, username) + user = utils.get_single_record(session, alchemy.User, 'User', username) if user is None: raise CREDENTIALS_EXCEPTION diff --git a/app/utils/__init__.py b/app/utils/__init__.py index 9b1fba5..e4b8c45 100644 --- a/app/utils/__init__.py +++ b/app/utils/__init__.py @@ -9,14 +9,14 @@ def make_primary_annotation(name=str): return Annotated[str, Path(description=f'{name}, either id (int) or code')] -def get_single_record(session, cls, code: str): +def get_single_record(session, cls, name: str, code: str): result = session.get(cls, code) if result is None: result = session.scalar(select(cls).where(cls.code == code)) if result is None: raise HTTPException( status_code=404, - detail=f"{cls.__class__.__name__} {code!r} not found.") + detail=f"{name} {code!r} not found.") return result