initial wip

.idea/runConfigurations/latest.xml

@@ -0,0 +1,13 @@
+<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/vpsx-db:latest" />
+        <option name="buildOnly" value="true" />
+        <option name="containerName" value="" />
+        <option name="sourceFilePath" value="Dockerfile" />
+      </settings>
+    </deployment>
+    <method v="2" />
+  </configuration>
+</component>

Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3
+
+LABEL authors="Bernhard Radermacher"
+
+RUN pip install mariadb pymysql sqlmodel uvicorn fastapi
+
+ENV DB_USER="must be set" \
+    DB_PASSWORD="must be set" \
+    DB_HOST="mariadb" \
+    DB_PORT=3306 \
+    DB_DATABASE="vpsx"
+
+COPY src /src
+
+COPY main.py /
+
+
+ENTRYPOINT ["/usr/local/bin/uvicorn", "main:app", "--host", "0.0.0.0"]
+

main.py

@@ -0,0 +1,33 @@
+import datetime
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from src.populate import populate
+import src.routers as routers
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    populate()
+    yield
+
+updated = datetime.datetime.now(tz=datetime.timezone.utc)
+
+app = FastAPI(lifespan=lifespan)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(routers.status)
+
+#
+# @app.get("/info/updated", tags=["info"])
+# async def get_timestamp_of_last_update():
+#     """Timestamp of last update (YYYY-MM-DD HH:MM:SS UTC)."""
+#     return updated.strftime("%Y-%m-%d %H:%M:%S")

pyproject.toml

@@ -0,0 +1,12 @@
+[project]
+name = "vpsx-db"
+version = "0.1.0"
+description = "Add your description here"
+requires-python = ">=3.13"
+dependencies = [
+    "fastapi>=0.116.1",
+    "mariadb>=1.1.13",
+    "pymysql>=1.1.2",
+    "sqlmodel>=0.0.24",
+    "uvicorn>=0.35.0",
+]

src/model/__init__.py

@@ -0,0 +1,4 @@
+from .contact import Contact
+from .status import Status
+
+from .util import get_session, count_rows

src/model/contact.py

@@ -0,0 +1,9 @@
+from sqlmodel import Field, SQLModel, Text
+
+
+class Contact(SQLModel, table=True):
+    id: int = Field(primary_key=True)
+    address: str = Field(max_length=256, unique=True)
+    description: str = Field(max_length=512)
+    notes: Text
+    status_id: str = Field(foreign_key="status.id")

src/model/status.py

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

src/model/util.py

@@ -0,0 +1,29 @@
+from sqlalchemy import func, select, URL
+
+import os
+
+from sqlmodel import create_engine, SQLModel, Session
+
+
+def get_engine():
+    engine_url = 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'},
+    )
+    engine = create_engine(engine_url)
+    SQLModel.metadata.create_all(engine)
+    return engine
+
+engine = get_engine()
+
+def get_session():
+    with Session(engine) as session:
+        yield session
+
+def count_rows(session, cls) -> int:
+    return session.execute(select(func.count()).select_from(cls)).scalar()

src/populate.py

@@ -0,0 +1,37 @@
+from sqlmodel import Session
+
+from .model.util import engine
+from src.model import count_rows, Status, Contact
+
+
+def populate_contact(session):
+    if count_rows(session, Contact) == 0:
+        for kwargs in (
+                dict(address="Active"),
+                dict(id="I", name="Inactive"),
+                dict(id="N", name="New"),
+                dict(id="P", name="Prepared"),
+                dict(id="X", name="eXcluded"),
+        ):
+            session.add(Status(**kwargs))
+
+
+
+def populate_status(session):
+    if count_rows(session, Status) == 0:
+        for kwargs in (
+                dict(id="A", name="Active"),
+                dict(id="I", name="Inactive"),
+                dict(id="N", name="New"),
+                dict(id="P", name="Prepared"),
+                dict(id="X", name="eXcluded"),
+        ):
+            session.add(Status(**kwargs))
+
+
+def populate():
+    with Session(engine) as session:
+
+        populate_status(session)
+
+        session.commit()

src/routers/__init__.py

@@ -0,0 +1 @@
+from .status import router as status

src/routers/status.py

@@ -0,0 +1,26 @@
+
+
+from fastapi import APIRouter, Depends, HTTPException, Path, Query
+from sqlmodel import select
+
+from ..model import Status, get_session
+
+router = APIRouter(
+    prefix="/status",
+    tags=["status"],
+)
+
+@router.get("/", response_model=list[Status])
+async def get_statuses(session=Depends(get_session)):
+    """List of Statuses"""
+    return session.exec(select(Status)).all()
+
+@router.get("/{status}", response_model=Status, responses={404: {"description": "Not found"}})
+async def get_status(
+        status: str,
+        session=Depends(get_session)):
+    """Get a Status"""
+    result = session.get(Status, status)
+    if not result:
+        raise HTTPException(status_code=404, detail=f"Status {status!r} not found")
+    return result

test_main.http

@@ -0,0 +1,26 @@
+# Test your FastAPI endpoints
+
+GET http://localhost:8888/info/updated
+Accept: application/json
+
+###
+
+#GET http://localhost:8888/calendar
+#Accept: application/json
+#
+####
+#
+#GET http://localhost:8888/calendar/1/dates/
+#Accept: application/json
+#
+####
+#
+#GET http://localhost:8888/calendar/1/dates?offset=1
+#Accept: application/json
+#
+####
+#
+#GET http://localhost:8888/calendar/1/dates?offset=-1
+#Accept: application/json
+#
+####