wip

.idea/runConfigurations/latest.xml

@@ -0,0 +1,12 @@
+<component name="ProjectRunConfigurationManager">
+  <configuration default="false" name="latest" type="docker-deploy" factoryName="dockerfile" server-name="Docker">
+    <deployment type="dockerfile">
+      <settings>
+        <option name="imageTag" value="docker.ctmapp.kiongroup.net/oms-db:latest" />
+        <option name="buildOnly" value="true" />
+        <option name="sourceFilePath" value="Dockerfile" />
+      </settings>
+    </deployment>
+    <method v="2" />
+  </configuration>
+</component>

.idea/runConfigurations/sandboxapi.xml

@@ -2,12 +2,19 @@
   <configuration default="false" name="sandboxapi" type="Python.FastAPI">
     <option name="additionalOptions" value="--reload --host 0.0.0.0 --port 1234" />
     <option name="file" value="$PROJECT_DIR$/app/main.py" />
-    <module name="sandboxapi" />
+    <module name="vpsx-fast" />
     <option name="ENV_FILES" value="" />
     <option name="INTERPRETER_OPTIONS" value="" />
     <option name="PARENT_ENVS" value="true" />
+    <envs>
+      <env name="DB_HOST" value="mariadb.ctmapp.kiongroup.net" />
+      <env name="DB_PASSWORD" value="fast" />
+      <env name="DB_PORT" value="3306" />
+      <env name="DB_USER" value="fast" />
+      <env name="DB_DATABASE" value="fast_vpsx" />
+    </envs>
     <option name="SDK_HOME" value="$PROJECT_DIR$/.venv/bin/python" />
-    <option name="SDK_NAME" value="uv (sandboxapi)" />
+    <option name="SDK_NAME" value="uv (vpsx-fast)" />
     <option name="WORKING_DIRECTORY" value="" />
     <option name="IS_MODULE_SDK" value="false" />
     <option name="ADD_CONTENT_ROOTS" value="true" />

Dockerfile

@@ -0,0 +1,23 @@
+FROM python:3
+
+LABEL authors="Bernhard Radermacher"
+
+RUN pip install fastapi mariadb pymysql passlib[bcrypt] pyjwt python-multipart sqlmodel uvicorn ldap3
+
+ARG DB_USER="must be set"
+ARG DB_PASSWORD="must be set"
+
+ENV DB_USER=$DB_USER \
+    DB_PASSWORD=$DB_PASSWORD \
+    DB_HOST="mariadb.ctmapp.kiongroup.net" \
+    DB_PORT=3306 \
+    DB_DATABASE="fast_vpsx" \
+    LOG_LEVEL="INFO"
+
+COPY app /
+
+
+#ENTRYPOINT ["python", "main.py"]
+#ENTRYPOINT ["/bin/bash"]
+
+ENTRYPOINT ["/usr/local/bin/uvicorn", "main:app", "--host", "0.0.0.0"]

app/alchemy/base.plantuml

@@ -0,0 +1,24 @@
+@startuml
+
+
+top to bottom direction
+skinparam linetype ortho
+
+package base {
+
+  abstract Base {
+    + id\t\t\t\tInteger\t\t<<PK>>  {field}
+    # __tablename__\tstr   {field}
+  }
+
+  object to_snake_case {
+    name\tstr
+  }
+
+  object bidirectional_relationship {
+    cls\t\t\t\tclass
+    foreign_table_cls\tclass
+  }
+
+ }
+@enduml

app/alchemy/base.py

@@ -37,7 +37,3 @@ class Base(DeclarativeBase):
         return to_snake_case(cls.__name__)
 
 
-# noinspection PyPep8Naming
-
-
-

app/alchemy/contact.plantuml

@@ -0,0 +1,27 @@
+@startuml
+
+'!include base.plantuml
+!include status.plantuml
+!include user.plantuml
+
+skinparam linetype polyline
+
+
+class Contact {
+  {field}  + code\tString(80)\t<<Unique>>
+  {field}  + address\tString(255) | None
+  {field}  + notes\tText | None
+}
+
+abstract ContactForeignKey {
+    contact_id\tContact.id {field}
+  + contact\t\tContact      {field}
+}
+
+
+'Base              <|--Contact
+StatusForeignKey  <|-[#blue]- Contact
+Versioned         <|-[#blue]- Contact
+Contact *-- ContactForeignKey
+
+@enduml

app/alchemy/contact.puml

@@ -0,0 +1,25 @@
+@startuml
+
+'!include base.plantuml
+!include status.puml
+!include user.puml
+
+skinparam linetype polyline
+
+entity contact {
+  * id\t\t\tint
+  --
+  * code\t\tvarchar(80)\t<<UK>>
+    address\t\tvarchar(255)
+    notes\t\ttext
+    status_id\t\tvarchar(3)\t<<FK>>
+  * _created__\tdatetime
+    _updated__\tdatetime
+    _user__id\t\tint\t\t\t<<FK>>
+}
+
+status --{ contact : status_id
+user   ..{ contact : _user__id
+
+
+@enduml

app/alchemy/contact.py

@@ -13,7 +13,7 @@ class Contact(StatusForeignKey, Versioned, Base):
     """Contact"""
 
     code: Mapped[str] = mapped_column(String(80), unique=True)
-    address: Mapped[str | None] = mapped_column(String(253))
+    address: Mapped[str | None] = mapped_column(String(255))
     notes: Mapped[str | None] = mapped_column(Text)
 
 

app/alchemy/location.plantuml

@@ -0,0 +1,72 @@
+@startuml
+
+'!include base.plantuml
+!include status.plantuml
+!include contact.plantuml
+
+skinparam linetype polyline
+
+
+class Country {
+  {field}  + code\t\tString(2)\t\t<<Unique>>
+  {field}  + name\t\tString(80)
+  {field}  + notes\t\tText | None
+           +  __repr__()\t\tstr
+}
+
+abstract CountryForeignKey {
+     country_id\tCountry.id     {field}
+   + country\t\tCountry         {field}
+}
+
+
+class LocationCode {
+  {field}  + code\t\tString(8)\t\t<<Unique>>
+  {field}  + description\tString(80)
+  {field}  + notes\t\tText | None
+           +  __repr__()\t\tstr
+}
+
+abstract LocationCodeForeignKey {
+    location_code_id\tLocationCode.id
+  + location_code\t\tLocationCode
+}
+
+
+class Location {
+  {field}  + code\t\tString(30)
+  {field}  + description\tString(80)
+  {field}  + notes\t\tText | None
+  {field}    <<Unique>>\n\tlocation_code_id\n\tcode
+           +  __repr__()\t\tstr
+}
+
+abstract LocationForeignKey {
+    location_id\tLocation.id
+  + location\t\tLocation
+}
+
+
+
+
+
+
+Versioned         <|-[#blue]-  Country
+StatusForeignKey  <|-[#blue]-  Country
+Country           *--          CountryForeignKey
+
+
+Versioned         <|-[#blue]-  LocationCode
+StatusForeignKey  <|-[#blue]-  LocationCode
+LocationCode      *--          LocationCodeForeignKey
+CountryForeignKey <|-[#blue]-  LocationCode
+ContactForeignKey o..          LocationCode
+
+
+Versioned         <|-[#blue]-  Location
+StatusForeignKey  <|-[#blue]-  Location
+Location          *--          LocationForeignKey
+LocationCodeForeignKey <|-[#blue]-  Location
+ContactForeignKey o..          Location
+
+@enduml

app/alchemy/location.puml

@@ -0,0 +1,66 @@
+@startuml
+
+'!include base.plantuml
+!include status.puml
+!include contact.puml
+
+skinparam linetype polyline
+
+
+entity country {
+  * id\t\t\tint
+  --
+  * code\t\tvarchar(2)\t<<UK>>
+    name\t\tvarchar(80)
+    notes\t\ttext
+    status_id\t\tvarchar(3)\t<<FK>>
+  * _created__\tdatetime
+    _updated__\tdatetime
+    _user__id\t\tint\t\t\t<<FK>>
+}
+
+
+
+entity location_code {
+  * id\t\t\tint
+  --
+  * code\t\tvarchar(8)\t<<UK>>
+    description\tvarchar(80)
+    notes\t\ttext
+  * country_id\tint\t\t\t<<FK>>
+    contact_id\tint\t\t\t<<FK>>
+    status_id\t\tvarchar(3)\t<<FK>>
+  * _created__\tdatetime
+    _updated__\tdatetime
+    _user__id\t\tint\t\t\t<<FK>>
+}
+
+
+entity location {
+  * id\t\t\t\tint
+  .. <<UK>> ..
+  * location_code_id\tint\t\t\t<<FK>>
+  * code\t\t\tvarchar(30)
+  --
+    description\t\tvarchar(80)
+    notes\t\t\ttext
+    contact_id\tint\t\t\t<<FK>>
+    status_id\t\t\tvarchar(3)\t<<FK>>
+  * _created__\t\tdatetime
+    _updated__\t\tdatetime
+    _user__id\t\t\tint\t\t\t<<FK>>
+}
+
+contact       ..{ location      : contact_id
+contact       ..{ location_code : contact_id
+country       --{ location_code : country_id
+location_code --{ location      : location_code_id
+status        --{ country       : status_id
+status        --{ location      : status_id
+status        --{ location_code : status_id
+user          ..{ country       : _user__id
+user          ..{ location      : _user__id
+user          ..{ location_code : _user__id
+
+
+@enduml

app/alchemy/status.plantuml

@@ -0,0 +1,20 @@
+@startuml
+
+'!include base.plantuml
+
+
+class Status {
+    id\t\tString(3)\t\t<<Primary>>  {field}
+  + name\tString(30)\t<<Unique>>    {field}
+}
+
+'Base <|-- Status
+
+abstract StatusForeignKey {
+    status_id\t\tStatus.id {field}
+  + status\t\tStatus      {field}
+}
+
+Status *-- StatusForeignKey
+
+@enduml

app/alchemy/status.puml

@@ -0,0 +1,9 @@
+@startuml
+
+entity status {
+  * id\t\tvarchar(3)
+  --
+  * name\tvarchar(30)\t<<UK>>
+}
+
+@enduml

app/alchemy/user.plantuml

@@ -0,0 +1,31 @@
+@startuml
+
+'!include base.plantuml
+!include status.plantuml
+skinparam linetype ortho
+
+
+class User {
+  {field}  + code\t\tString(253)\t<<Unique>>
+  {field}  + name\t\tString(253)
+  {field}  + password\tString(255) | None
+  {field}  + ldap_name\tString(255) | None
+  {field}  + notes\t\tText | None
+           + __repr__()\t\tstr
+}
+
+abstract Versioned {
+  {field}  + user__\t\tUser
+  {field}  # _created__\tDatetime
+  {field}  # _updated__\tDatetime | None
+  {field}    _user__id\t\tInteger | None
+  {field}  # __versioned__\t= True
+}
+
+'Base <|-- User
+StatusForeignKey <|-[#blue]- User
+Versioned <|-[#blue]- User
+User *-- Versioned
+
+
+@enduml

app/alchemy/user.puml

@@ -0,0 +1,25 @@
+@startuml
+
+
+!include status.puml
+skinparam linetype polyline
+
+
+entity user {
+  * id\t\t\tint
+  --
+  * code\t\tvarchar(255)\t<<UK>>
+  * name\t\tvarchar(255)
+    password\tvarchar(255)
+    ldap_name\tvarchar(255)
+    notes\t\ttext
+    status_id\t\tvarchar(3)\t<<FK>>
+  * _created__\tdatetime
+    _updated__\tdatetime
+    _user__id\t\tint\t\t\t<<FK>>
+}
+
+status --{ user : status_id
+user   ..{ user : _user__id
+
+@enduml

app/alchemy/user.py

@@ -61,15 +61,15 @@ class Versioned:
 
     # noinspection PyMethodParameters
     @declared_attr
-    def _user__(cls) -> Mapped["User"]:
+    def user__(cls) -> Mapped["User"]:
         return relationship()
 
 
 class User(StatusForeignKey, Versioned, Base):
     """User"""
 
-    code: Mapped[str] = mapped_column(String(253), unique=True)
-    name: Mapped[str] = mapped_column(String(253))
+    code: Mapped[str] = mapped_column(String(255), unique=True)
+    name: Mapped[str] = mapped_column(String(255))
     password: Mapped[str | None] = mapped_column(String(255))
     ldap_name: Mapped[str | None] = mapped_column(String(255))
     notes: Mapped[str | None] = mapped_column(Text)
@@ -90,10 +90,8 @@ def initialize_user(target, connection, **kwargs):
         session.commit()
         for kwargs in (
                 dict(code="CTM", name="Control-M", password=get_password_hash("secret"), notes="user for automation"),
-                dict(code="exde37c8", name="Bernhard Radermacher",
-                     password=get_password_hash("secret"),
-                     ldap_name="a0061806@kiongroup.com",
-                     ),
+                dict(code="de31c7", name="Andy Le", ldap_name="a0032514@kiongroup.com"),
+                dict(code="exde37c8", name="Bernhard Radermacher", ldap_name="a0061806@kiongroup.com"),
         ):
             kwargs.update(dict(status_id='A', _user__id=qsys.id))
             session.add(User(**kwargs))

app/dependencies/engine.py

@@ -1,11 +1,22 @@
-from sqlalchemy import create_engine
+import os
+import sqlalchemy
 
-# engine_url="sqlite+pysqlite:///vpsx.db"
-engine_url="mariadb+pymysql://fast:fast@localhost/fast_vpsx?charset=utf8mb4"
+# # engine_url="sqlite+pysqlite:///vpsx.db"
+# engine_url="mariadb+pymysql://fast:fast@mariadb.ctmapp.kiongroup.net/fast_vpsx?charset=utf8mb4"
+
+engine_url = sqlalchemy.URL.create(
+    drivername="mariadb+pymysql",
+    username=os.getenv("DB_USER"),
+    password=os.getenv("DB_PASSWORD"),
+    host=os.getenv("DB_HOST"),
+    port=int(os.getenv("DB_PORT")),
+    database=os.getenv("DB_DATABASE"),
+    query={'charset': 'utf8mb4'},
+)
 
 
 def get_engine():
-    return create_engine(engine_url)
+    return sqlalchemy.create_engine(engine_url)
 
 engine = get_engine()
 

app/main.py

@@ -1,15 +1,13 @@
 import inspect
 from contextlib import asynccontextmanager
-from typing import Annotated
 
 import fastapi
-from fastapi.security import OAuth2PasswordRequestForm
-
-from app.alchemy import Base
-from app.dependencies import engine
-from fastapi import FastAPI, Depends, HTTPException
+from fastapi import FastAPI
 from starlette.middleware.cors import CORSMiddleware
-from app import routers
+
+from alchemy import Base
+import routers
+from dependencies import engine
 
 
 @asynccontextmanager
@@ -17,15 +15,10 @@ async def lifespan(app: FastAPI):
     Base.metadata.create_all(engine)
     yield
 
+
 app = FastAPI(lifespan=lifespan)
 app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"])
 
-
-# app.include_router(routers.contact)
-# app.include_router(routers.country)
-# app.include_router(routers.status)
-# app.include_router(routers.user)
-
 for i in inspect.getmembers(routers):
     if isinstance(i[1], fastapi.routing.APIRouter):
         app.include_router(i[1])

app/routers/contact.py

@@ -9,118 +9,94 @@ 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
+from .models import Contact, ContactCreate, ContactUpdate
+
 
 PRIMARY_ANNOTATION = utils.make_primary_annotation('Contact')
 
-
-# ----------------------------------------------------------------
-# Models
-
-class ContactBase(SQLModel):
-    code: str = Field(max_length=80, unique=True)
-    address: str = Field(max_length=253)
-    notes: str | None
-
-
-class Contact(ContactBase):
-    id: int | None = Field(default=None, primary_key=True)
-
-
-class ContactCreate(ContactBase):
-    address: str | None = None
-    notes: str | None = None
-
-
-class ContactPublic(ContactBase):
-    id: int
-    status: Status
-
-
-class ContactUpdate(ContactBase):
-    code: str | None = None
-    address: str | None = None
-    notes: str | None = None
-
-
 # ----------------------------------------------------------------
 # Routes
 
 router = APIRouter(prefix="/contact", tags=["contact"])
 
 
-@router.get("/", response_model=list[ContactPublic])
+@router.get("/",
+            response_model=list[Contact])
 async def get_contacts(
         offset: int = 0,
         limit: Annotated[int, Query] = 100,
         session=Depends(get_session)):
-    """Get list of all contacts"""
-    if limit < 1:
-        limit = sys.maxsize
-    return session.exec(select(alchemy.Contact).offset(offset).limit(limit)).all()
+    """Get list of all Contacts"""
+    return session.exec(
+        select(
+            alchemy.Contact
+        ).offset(offset).limit(limit or sys.maxsize)).all()
 
 
-@router.get("/{contact_id}",
-            response_model=ContactPublic,
+# noinspection PyTypeHints
+@router.get("/{contact}",
+            response_model=Contact,
             responses={404: {"description": "Not found"}})
 async def get_contact(
-        contact_id: PRIMARY_ANNOTATION,
+        contact: PRIMARY_ANNOTATION,
         session=Depends(get_session)):
-    return utils.get_single_record(session, alchemy.Contact, contact_id)
+    return utils.get_single_record(session, alchemy.Contact, 'Contact', contact)
 
 
+# noinspection PyTypeHints
 @router.post("/",
-             response_model=ContactPublic)
+             response_model=Contact)
 async def create_contact(
-        contact: ContactCreate,
+        data: ContactCreate,
         current_user: ACTIVE_USER,
         session=Depends(get_session)):
     return create_item(
         session=session,
         cls=alchemy.Contact,
+        name='Contact',
         current_user=current_user,
-        data=Contact.model_validate(contact))
+        data=data)
 
 
-@router.patch("/{contact_id}",
-              response_model=ContactPublic,
+@router.patch("/{contact}",
+              response_model=Contact,
               responses={404: {"description": "Not found"}})
 async def update_contact(
-        contact_id: PRIMARY_ANNOTATION,
-        contact: ContactUpdate,
+        contact: PRIMARY_ANNOTATION,
+        data: ContactUpdate,
         current_user: ACTIVE_USER,
         session=Depends(get_session)):
     return update_item(
         session=session,
         current_user=current_user,
-        item=utils.get_single_record(session, alchemy.Contact, contact_id),
-        data=contact)
+        item=utils.get_single_record(session, alchemy.Contact, 'Contact', contact),
+        data=data)
 
 
-@router.put("/{contact_id}/activate",
-            response_model=ContactPublic,
-              responses={404: {"description": "Not found"}})
+@router.put("/{contact}/activate",
+            response_model=Contact,
+            responses={404: {"description": "Not found"}})
 async def activate_contact(
-        contact_id: PRIMARY_ANNOTATION,
+        contact: 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.Contact, contact_id),
+        item=utils.get_single_record(session, alchemy.Contact, 'Contact', contact),
         status='A')
 
 
-@router.put("/{contact_id}/deactivate",
-            response_model=ContactPublic,
-              responses={404: {"description": "Not found"}})
+@router.put("/{contact}/deactivate",
+            response_model=Contact,
+            responses={404: {"description": "Not found"}})
 async def deactivate_contact(
-        contact_id: PRIMARY_ANNOTATION,
+        contact: 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.Contact, contact_id),
+        item=utils.get_single_record(session, alchemy.Contact, 'Contact', contact),
         status='I')

app/routers/country.py

@@ -1,125 +0,0 @@
-import sys
-from typing import Annotated
-
-from fastapi import APIRouter, Query, Depends
-from sqlmodel import SQLModel, Field
-from sqlmodel 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 CountryBase(SQLModel):
-    code: str = Field(max_length=2, unique=True)
-    name: str = Field(max_length=80)
-    notes: str | None
-
-
-class Country(CountryBase):
-    id: int | None = Field(default=None, primary_key=True)
-
-
-class CountryCreate(CountryBase):
-    notes: str | None = None
-
-
-class CountryPublic(CountryBase):
-    id: int
-    status: Status
-
-
-class CountryUpdate(CountryBase):
-    code: str | None = None
-    name: str | None = None
-    notes: str | None = None
-
-
-# ----------------------------------------------------------------
-# Routes
-
-router = APIRouter(prefix="/country", tags=["country"])
-
-
-@router.get("/", response_model=list[CountryPublic])
-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
-    return session.exec(select(alchemy.Country).offset(offset).limit(limit)).all()
-
-
-@router.get("/{country_id}",
-            response_model=CountryPublic,
-            responses={404: {"description": "Not found"}})
-async def get_country(
-        country_id: PRIMARY_ANNOTATION,
-        session=Depends(get_session)):
-    return utils.get_single_record(session, alchemy.Country, country_id)
-
-
-@router.post("/",
-             response_model=CountryPublic)
-async def create_country(
-        country: CountryCreate,
-        current_user: ACTIVE_USER,
-        session=Depends(get_session)):
-    return create_item(
-        session=session,
-        cls=alchemy.Country,
-        current_user=current_user,
-        data=Country.model_validate(country))
-
-
-@router.patch("/{country_id}",
-              response_model=CountryPublic,
-              responses={404: {"description": "Not found"}})
-async def update_country(
-        country_id: PRIMARY_ANNOTATION,
-        country: 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_id),
-        data=country)
-
-
-@router.put("/{country_id}/activate",
-            response_model=CountryPublic,
-              responses={404: {"description": "Not found"}})
-async def activate_country(
-        country_id: 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_id),
-        status='A')
-
-
-@router.put("/{country_id}/deactivate",
-            response_model=CountryPublic,
-              responses={404: {"description": "Not found"}})
-async def deactivate_country(
-        country_id: 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_id),
-        status='I')

app/routers/location.py

@@ -4,14 +4,13 @@ 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 utils import update_item
+from .models.location import LocationCode, Country, LocationCreate, LocationCodeCreate, CountryCreate, CountryUpdate
 from .user import ACTIVE_USER
 
 PRIMARY_ANNOTATION = utils.make_primary_annotation('Country')
@@ -73,66 +72,6 @@ PRIMARY_ANNOTATION = utils.make_primary_annotation('Country')
 #     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
@@ -168,7 +107,7 @@ async def get_country(
 
 
 @router.post("/",
-              response_model=Country)
+             response_model=Country)
 async def create_country(
         country: CountryCreate,
         current_user: ACTIVE_USER,
@@ -243,7 +182,7 @@ async def deactivate_country(
 
 
 @router.post("/{country}",
-            response_model=LocationCode,)
+             response_model=LocationCode, )
 async def create_location_code(
         country: PRIMARY_ANNOTATION,
         data: LocationCodeCreate,
@@ -274,7 +213,7 @@ async def create_location_code(
 
 
 @router.post("/{country}/{location_code}",
-            response_model=LocationCode,)
+             response_model=LocationCode, )
 async def create_location(
         country: PRIMARY_ANNOTATION,
         location_code: PRIMARY_ANNOTATION,

app/routers/location_code.py

@@ -1,134 +0,0 @@
-import sys
-from typing import Annotated
-
-from fastapi import APIRouter, Query, Depends
-from sqlmodel import SQLModel, Field
-from sqlmodel 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
-from .contact import Contact
-from .country import Country
-
-
-PRIMARY_ANNOTATION = utils.make_primary_annotation('Location Code')
-
-
-# ----------------------------------------------------------------
-# Models
-
-
-class LocationCodeBase(SQLModel):
-    code: str = Field(max_length=8, unique=True)
-    description: str = Field(max_length=256)
-    notes: str | None
-
-
-class LocationCode(LocationCodeBase):
-    id: int | None = Field(default=None, primary_key=True)
-    country: Country
-    contact: Contact | None = None
-
-
-class LocationCodeCreate(LocationCodeBase):
-    country_id: int
-    contact_id : int | None = None
-    notes: str | None = None
-
-
-class LocationCodePublic(LocationCode):
-    status: Status
-
-
-class LocationCodeUpdate(LocationCodeBase):
-    code: str | None = None
-    description: str | None = None
-    country_id: int | None = None
-    contact_id: int | None = None
-    notes: str | None = None
-
-
-# ----------------------------------------------------------------
-# Routes
-
-router = APIRouter(prefix="/location_code", tags=["location_code"])
-
-
-@router.get("/", response_model=list[LocationCodePublic])
-async def get_location_codes(
-        offset: int = 0,
-        limit: Annotated[int, Query] = 100,
-        session=Depends(get_session)):
-    """Get list of all location codes"""
-    if limit < 1:
-        limit = sys.maxsize
-    return session.exec(select(alchemy.LocationCode).offset(offset).limit(limit)).all()
-
-
-@router.get("/{location_code_id}",
-            response_model=LocationCodePublic,
-            responses={404: {"description": "Not found"}})
-async def get_location_code(
-        location_code_id: PRIMARY_ANNOTATION,
-        session=Depends(get_session)):
-    return utils.get_single_record(session, alchemy.LocationCode, location_code_id)
-
-
-@router.post("/",
-             response_model=LocationCodePublic)
-async def create_location_code(
-        location_code: LocationCodeCreate,
-        current_user: ACTIVE_USER,
-        session=Depends(get_session)):
-    return create_item(
-        session=session,
-        cls=alchemy.LocationCode,
-        current_user=current_user,
-        data=LocationCodeCreate.model_validate(location_code))
-
-
-@router.patch("/{location_code_id}",
-              response_model=LocationCodePublic,
-              responses={404: {"description": "Not found"}})
-async def update_location_code(
-        location_code_id: PRIMARY_ANNOTATION,
-        location_code: LocationCodeUpdate,
-        current_user: ACTIVE_USER,
-        session=Depends(get_session)):
-    return update_item(
-        session=session,
-        current_user=current_user,
-        item=utils.get_single_record(session, alchemy.LocationCode, location_code_id),
-        data=location_code)
-
-
-@router.put("/{location_code_id}/activate",
-            response_model=LocationCodePublic,
-              responses={404: {"description": "Not found"}})
-async def activate_location_code(
-        locationCode_id: 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.LocationCode, locationCode_id),
-        status='A')
-
-
-@router.put("/{location_code_id}/deactivate",
-            response_model=LocationCodePublic,
-              responses={404: {"description": "Not found"}})
-async def deactivate_location_code(
-        location_code_id: 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.LocationCode, location_code_id),
-        status='I')

app/routers/models/__init__.py

@@ -0,0 +1,5 @@
+from .contact import Contact, ContactCreate, ContactUpdate
+from .location import Country, CountryCreate, CountryUpdate, LocationCode, LocationCodeCreate, Location, LocationCreate
+from .status import Status
+
+

app/routers/models/contact.py

@@ -0,0 +1,23 @@
+from typing import Annotated
+
+from pydantic import BaseModel, Field
+
+from .status import Status
+
+
+class Contact(BaseModel):
+    id: int
+    code: str
+    address: str | None
+    notes: str | None
+    status: Status | None
+
+
+class ContactCreate(BaseModel):
+    code: str = Field(max_length=80)
+    address: Annotated[str | None, Field(max_length=253)] = None
+    notes: str | None = None
+
+
+class ContactUpdate(ContactCreate):
+    code: Annotated[str | None, Field(max_length=80)] = None

app/routers/models/location.py

@@ -0,0 +1,66 @@
+from typing import Annotated
+
+from pydantic import BaseModel, Field
+
+from .contact import Contact
+from .status import Status
+
+
+class Location(BaseModel):
+    id: int
+    code: str
+    description: str | None
+    notes: str | None
+    status: Status
+    contact: Contact | None
+
+class LocationCreate(BaseModel):
+    code: str = Field(max_length=30)
+    description: Annotated[str | None, Field(max_length=80)] = None
+    notes: str | None = None
+
+class LocationUpdate(LocationCreate):
+    code: Annotated[str | None, Field(max_length=30)] = None
+
+
+class LocationCode(BaseModel):
+    id: int
+    code: str
+    description: str | None
+    notes: str | None
+    status: Status
+    locations: list[Location]
+    contact: Contact | None
+
+class LocationCodeCreate(BaseModel):
+    code: str = Field(max_length=8)
+    description: Annotated[str | None, Field(max_length=80)] = None
+    notes: str | None = None
+    locations: list[LocationCreate] | None = None
+
+
+
+class Country(BaseModel):
+    id: int
+    code: str
+    name: str
+    notes: str | None
+    status: Status
+    location_codes: list[LocationCode] | 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

app/routers/models/status.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class Status(BaseModel):
+    id: str
+    name: str

app/routers/status.py

@@ -3,7 +3,7 @@ from sqlmodel import select, Session
 
 import alchemy
 from dependencies import get_session
-from routers.status_model import Status
+from routers.models import Status
 
 router = APIRouter(prefix="/status", tags=["status"])
 

app/routers/status_model.py

@@ -1,6 +0,0 @@
-from sqlmodel import SQLModel, Field
-
-
-class Status(SQLModel):
-    id: str | None = Field(max_length=3, primary_key=True)
-    name: str = Field(max_length=30, unique=True)

app/routers/user.py

@@ -16,7 +16,7 @@ from sqlmodel import SQLModel, Field, Session, select
 import alchemy
 import utils
 from dependencies import get_session
-from routers.status_model import Status
+from routers.models import Status
 
 logging.getLogger('passlib').setLevel(logging.ERROR)
 

app/utils/__init__.py

@@ -35,11 +35,12 @@ def set_item_status(session, current_user, item, status: str):
 
 def update_item(session, current_user, item, data):
     for k, v in data.model_dump(exclude_unset=True).items():
-        if (k == 'code' and
-                session.scalar(select(item.__class__).where(item.__class__.code == v)) != item):
-            raise HTTPException(status_code=422,
-                                detail=[dict(msg=f"{item.__class__.__name__} {k!r} already exists",
-                                             type="Integrity Error")])
+        if k == 'code':
+            r = session.scalar(select(item.__class__).where(item.__class__.code == v))
+            if r and r != item:
+                raise HTTPException(status_code=422,
+                                    detail=[dict(msg=f"{item.__class__.__name__} {v!r} already exists",
+                                                 type="Integrity Error")])
         setattr(item, k, v)
     item._user__id = current_user.id
     try:
@@ -52,11 +53,11 @@ def update_item(session, current_user, item, data):
     return item
 
 
-def create_item(session, cls, current_user, data):
+def create_item(session, cls, name, current_user, data):
     item = cls(**data.model_dump())
     if session.scalar(select(cls).where(cls.code == item.code)):
         raise HTTPException(status_code=422,
-                            detail=[dict(msg=f"{cls.__class__.__name__} {item.code} already exists",
+                            detail=[dict(msg=f"{name} {item.code!r} already exists",
                                          type="Integrity Error")])
     item._user__id = current_user.id
     session.add(item)

doc/process/business/add_user.adoc

@@ -0,0 +1,27 @@
+= Add User
+:toc:
+:icons: font
+
+== Purpose
+
+Add user to system.
+
+== Required Information
+
+[%autowidth]
+[cols="l,a"]
+|===
+|Item |Description
+
+|code
+|Identifies the user, sometimes called _user id_ or _username_. This _should_ be the same as the respective *Windows Domain User ID*
+
+|name
+|Name of user for display.
+
+|ldap_name
+|Identity of user in LDAP to allow logon using *Windows Domain Password*.
+
+|notes
+|(optional) Internal notes.
+|===