CRUD API, CLI, tests, and docker compose

API.md

@@ -0,0 +1,122 @@
+CRUD service for tracking the state of servers across multiple datacenters.
+
+The project is implemented as a small production-style backend service with:
+- FastAPI
+- PostgreSQL
+- Raw SQL (psycopg)
+- Docker Compose
+- CLI tool
+- pytest test suite (including Docker-based execution)
+
+---
+
+## How to run the application
+
+### Requirements
+- Docker
+- Docker Compose
+
+From the project root directory:
+
+```bash
+docker compose up --build
+
+
+After startup the API will be available at:
+API base URL: http://localhost:8000
+Swagger / OpenAPI UI: http://localhost:8000/docs
+
+A server has the following fields:
+
+{
+  "id": 1,
+  "hostname": "srv-1",
+  "ip_address": "10.0.0.1",
+  "state": "active",
+  "datacenter": "berlin",
+  "created_at": "2026-01-18T12:17:01Z",
+  "updated_at": "2026-01-18T12:17:01Z"
+}
+
+Validation rules:
+hostname must be unique
+ip_address must be a valid IPv4 or IPv6 address
+
+state must be one of:
+active
+offline
+retired
+
+API endpoints
+Create a server
+curl -X POST http://localhost:8000/servers \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "hostname": "srv-1",
+    "ip_address": "10.0.0.1",
+    "state": "active",
+    "datacenter": "berlin"
+  }'
+
+List all servers
+curl http://localhost:8000/servers
+
+Get server by id
+curl http://localhost:8000/servers/1
+
+Update a server
+curl -X PUT http://localhost:8000/servers/1 \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "state": "offline"
+  }'
+
+Delete a server
+curl -X DELETE http://localhost:8000/servers/1
+
+The CLI communicates with the HTTP API.
+
+Optionally set the API URL:
+
+export API_URL=http://localhost:8000
+
+
+Examples:
+
+python cli.py create srv-2 10.0.0.2 active --datacenter ber
+python cli.py list
+python cli.py get 1
+python cli.py update 1 --state retired
+python cli.py delete 1
+
+Running tests (Docker)
+
+Tests can be executed entirely inside Docker, using the same image and dependencies
+as the API service.
+
+Start PostgreSQL
+docker compose up -d db
+
+Run tests in a container
+docker compose run --rm tests
+
+Expected output:
+
+6 passed
+
+Notes about test execution:
+Tests use the same PostgreSQL service defined in docker-compose.yml
+The test suite truncates the servers table between tests
+During negative tests (e.g. duplicate hostname), PostgreSQL logs
+unique constraint violation errors — this is expected and handled by the API
+Running tests locally (optional)
+
+If you prefer to run tests outside Docker:
+export DATABASE_URL=postgresql://inventory:inventory@localhost:5432/inventory
+pip install -r requirements.txt
+pytest -q
+
+Notes:
+Database schema is initialized automatically on application startup
+All database operations are implemented using raw SQL (no ORM)
+The CLI is a thin client that communicates with the API over HTTP

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 8000
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
+

app/db.py

@@ -0,0 +1,42 @@
+import os
+from contextlib import contextmanager
+from typing import Iterator
+
+import psycopg
+from psycopg.rows import dict_row
+
+DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://inventory:inventory@localhost:5432/inventory")
+
+
+@contextmanager
+def get_conn() -> Iterator[psycopg.Connection]:
+    conn = psycopg.connect(DATABASE_URL, row_factory=dict_row)
+    try:
+        yield conn
+        conn.commit()
+    except Exception:
+        conn.rollback()
+        raise
+    finally:
+        conn.close()
+
+
+def init_db() -> None:
+    ddl = """
+    CREATE TABLE IF NOT EXISTS servers (
+      id BIGSERIAL PRIMARY KEY,
+      hostname TEXT NOT NULL UNIQUE,
+      ip_address INET NOT NULL,
+      state TEXT NOT NULL,
+      datacenter TEXT,
+      created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+      updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+      CONSTRAINT servers_state_check CHECK (state IN ('active', 'offline', 'retired'))
+    );
+
+    CREATE INDEX IF NOT EXISTS idx_servers_state ON servers(state);
+    """
+    with get_conn() as conn:
+        with conn.cursor() as cur:
+            cur.execute(ddl)
+

app/main.py

@@ -0,0 +1,135 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, HTTPException, Response, status
+from psycopg.errors import CheckViolation, UniqueViolation
+
+from app.db import get_conn, init_db
+from app.schemas import ServerCreate, ServerOut, ServerUpdate
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Initialize DB schema on startup
+    init_db()
+    yield
+    # Nothing to cleanup here (connections are per-request)
+
+
+app = FastAPI(title="Server Inventory API", version="1.0.0", lifespan=lifespan)
+
+
+def _row_to_out(row: dict) -> ServerOut:
+    return ServerOut(
+        id=row["id"],
+        hostname=row["hostname"],
+        ip_address=str(row["ip_address"]),
+        state=row["state"],
+        datacenter=row.get("datacenter"),
+        created_at=row["created_at"].isoformat(),
+        updated_at=row["updated_at"].isoformat(),
+    )
+
+@app.post("/servers", response_model=ServerOut, status_code=status.HTTP_201_CREATED)
+def create_server(payload: ServerCreate) -> ServerOut:
+    sql = """
+    INSERT INTO servers (hostname, ip_address, state, datacenter)
+    VALUES (%s, %s, %s, %s)
+    RETURNING id, hostname, ip_address, state, datacenter, created_at, updated_at;
+    """
+    try:
+        with get_conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(sql, (payload.hostname, str(payload.ip_address), payload.state.value, payload.datacenter))
+                row = cur.fetchone()
+                return _row_to_out(row)
+    except UniqueViolation:
+        raise HTTPException(status_code=409, detail="hostname must be unique")
+    except CheckViolation:
+        # should be prevented by pydantic Enum, but keep DB safety
+        raise HTTPException(status_code=422, detail="invalid state")
+
+
+@app.get("/servers", response_model=list[ServerOut])
+def list_servers() -> list[ServerOut]:
+    sql = """
+    SELECT id, hostname, ip_address, state, datacenter, created_at, updated_at
+    FROM servers
+    ORDER BY id;
+    """
+    with get_conn() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql)
+            rows = cur.fetchall()
+            return [_row_to_out(r) for r in rows]
+
+
+@app.get("/servers/{server_id}", response_model=ServerOut)
+def get_server(server_id: int) -> ServerOut:
+    sql = """
+    SELECT id, hostname, ip_address, state, datacenter, created_at, updated_at
+    FROM servers
+    WHERE id = %s;
+    """
+    with get_conn() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, (server_id,))
+            row = cur.fetchone()
+            if not row:
+                raise HTTPException(status_code=404, detail="server not found")
+            return _row_to_out(row)
+
+
+@app.put("/servers/{server_id}", response_model=ServerOut)
+def update_server(server_id: int, payload: ServerUpdate) -> ServerOut:
+    # Build dynamic raw SQL safely with whitelist + parameter list
+    fields = []
+    params: list[object] = []
+
+    if payload.hostname is not None:
+        fields.append("hostname = %s")
+        params.append(payload.hostname)
+    if payload.ip_address is not None:
+        fields.append("ip_address = %s")
+        params.append(str(payload.ip_address))
+    if payload.state is not None:
+        fields.append("state = %s")
+        params.append(payload.state.value)
+    if payload.datacenter is not None:
+        fields.append("datacenter = %s")
+        params.append(payload.datacenter)
+
+    if not fields:
+        raise HTTPException(status_code=400, detail="no fields to update")
+
+    sql = f"""
+    UPDATE servers
+    SET {", ".join(fields)}, updated_at = now()
+    WHERE id = %s
+    RETURNING id, hostname, ip_address, state, datacenter, created_at, updated_at;
+    """
+    params.append(server_id)
+
+    try:
+        with get_conn() as conn:
+            with conn.cursor() as cur:
+                cur.execute(sql, tuple(params))
+                row = cur.fetchone()
+                if not row:
+                    raise HTTPException(status_code=404, detail="server not found")
+                return _row_to_out(row)
+    except UniqueViolation:
+        raise HTTPException(status_code=409, detail="hostname must be unique")
+    except CheckViolation:
+        raise HTTPException(status_code=422, detail="invalid state")
+
+
+@app.delete("/servers/{server_id}", status_code=status.HTTP_204_NO_CONTENT)
+def delete_server(server_id: int) -> Response:
+    sql = "DELETE FROM servers WHERE id = %s;"
+    with get_conn() as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql, (server_id,))
+            if cur.rowcount == 0:
+                raise HTTPException(status_code=404, detail="server not found")
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+

app/schemas.py

@@ -0,0 +1,35 @@
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field, IPvAnyAddress
+
+
+class ServerState(str, Enum):
+    active = "active"
+    offline = "offline"
+    retired = "retired"
+
+
+class ServerCreate(BaseModel):
+    hostname: str = Field(min_length=1, max_length=255)
+    ip_address: IPvAnyAddress
+    state: ServerState
+    datacenter: Optional[str] = Field(default=None, max_length=255)
+
+
+class ServerUpdate(BaseModel):
+    hostname: Optional[str] = Field(default=None, min_length=1, max_length=255)
+    ip_address: Optional[IPvAnyAddress] = None
+    state: Optional[ServerState] = None
+    datacenter: Optional[str] = Field(default=None, max_length=255)
+
+
+class ServerOut(BaseModel):
+    id: int
+    hostname: str
+    ip_address: str
+    state: ServerState
+    datacenter: Optional[str] = None
+    created_at: str
+    updated_at: str
+