kion/vpsx-fast
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

wip

Browse Source
This commit is contained in:
Bernhard Radermacher
2025-09-01 11:36:22 +00:00
parent eb1d8d793c
commit 292e296f01
26 changed files with 461 additions and 410 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
app/routers/contact.py
View File

@@ -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')

125
app/routers/country.py
View File

@@ -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')

71
app/routers/location.py
View File

@@ -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,

134
app/routers/location_code.py
View File

@@ -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')

5
app/routers/models/__init__.py Normal file
View File

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

23
app/routers/models/contact.py Normal file
View File

@@ -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

66
app/routers/models/location.py Normal file
View File

@@ -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

6
app/routers/models/status.py Normal file
View File

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

2
app/routers/status.py
View File

@@ -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"])

6
app/routers/status_model.py
View File

@@ -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)

2
app/routers/user.py
View File

@@ -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)