Bogdan Laban - Server inventory FastAPI app with CLI, tests, Docker

.gitignore

@@ -0,0 +1,12 @@
+__pycache__/
+*.py[cod]
+.Python
+env/
+venv/
+.venv/
+.mypy_cache/
+.pytest_cache/
+.DS_Store
+*.log
+/.idea/
+/.vscode/

API.md

@@ -0,0 +1,82 @@
+# Server Inventory
+
+CRUD API and CLI for managing servers across data centers. Built with FastAPI and PostgreSQL using raw SQL.
+
+## Quickstart (Docker Compose)
+
+```
+docker compose up --build
+```
+
+Services:
+- API at `http://localhost:8000`
+- PostgreSQL at `postgresql://postgres:postgres@localhost:5432/server_inventory`
+
+Environment variables:
+- `DATABASE_URL` (API) – defaults to `postgresql://postgres:postgres@localhost:5432/server_inventory`
+- `API_BASE_URL` (CLI) – defaults to `http://localhost:8000`
+
+## Running Locally (without Docker)
+
+```
+python -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+export DATABASE_URL="postgresql://postgres:postgres@localhost:5432/server_inventory"
+uvicorn app.main:app --reload
+```
+
+PostgreSQL must be running and reachable via `DATABASE_URL`. The app will create the `servers` table if it does not exist.
+
+## API Endpoints
+
+- `POST /servers` – create a server (unique `hostname`, valid IP, `state` in `active|offline|retired`)
+- `GET /servers` – list all servers
+- `GET /servers/{id}` – fetch one server
+- `PUT /servers/{id}` – update an existing server
+- `DELETE /servers/{id}` – delete a server
+
+Example request:
+
+```bash
+curl -X POST http://localhost:8000/servers \
+  -H "Content-Type: application/json" \
+  -d '{"hostname": "web-1", "ip_address": "10.0.0.1", "state": "active"}'
+```
+
+## CLI
+
+Commands run against the API (set `API_BASE_URL` if needed):
+
+```
+python -m cli.main list
+python -m cli.main get 1
+python -m cli.main create web-1 10.0.0.1 --state active
+python -m cli.main update 1 --hostname web-2 --state offline
+python -m cli.main delete 1
+```
+
+`update` merges provided fields with the current record so you do not need to pass all fields.
+
+## Testing
+
+Pytest requires access to PostgreSQL. By default it connects to `postgresql://postgres:postgres@localhost:5432/postgres`, creates a temporary database named `server_inventory_test`, and drops it afterwards. Override with:
+
+- `TEST_DATABASE_ADMIN_URL` – admin DSN to create/drop the test database
+- `TEST_DATABASE_NAME` – name of the temporary test database
+
+Run tests:
+
+```
+pytest
+```
+
+To test inside Docker, bring up the stack and run:
+
+```
+docker compose exec api env \
+  TEST_DATABASE_ADMIN_URL=postgresql://postgres:postgres@db:5432/postgres \
+  TEST_DATABASE_NAME=server_inventory_test \
+  pytest -q
+```
+

Dockerfile

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

app/__init__.py

@@ -0,0 +1 @@
+"""Application package for the server inventory API."""

app/db.py

@@ -0,0 +1,69 @@
+import os
+from typing import Iterable, Tuple
+
+from psycopg import Connection
+from psycopg.errors import UniqueViolation
+from psycopg_pool import ConnectionPool
+
+DEFAULT_DATABASE_URL = "postgresql://postgres:postgres@localhost:5432/server_inventory"
+ALLOWED_STATES: Tuple[str, ...] = ("active", "offline", "retired")
+
+
+def get_database_url() -> str:
+    return os.getenv("DATABASE_URL", DEFAULT_DATABASE_URL)
+
+
+def create_pool(database_url: str | None = None) -> ConnectionPool:
+    pool = ConnectionPool(
+        conninfo=database_url or get_database_url(),
+        min_size=1,
+        max_size=10,
+        timeout=10,
+        open=True,  # explicit to avoid future default change warnings
+    )
+    pool.wait()
+    return pool
+
+
+def ensure_schema(pool: ConnectionPool) -> None:
+    with pool.connection() as conn:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                CREATE TABLE IF NOT EXISTS servers (
+                    id SERIAL PRIMARY KEY,
+                    hostname TEXT NOT NULL UNIQUE,
+                    ip_address TEXT NOT NULL,
+                    state TEXT NOT NULL CHECK (state IN ('active', 'offline', 'retired')),
+                    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+                    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+                );
+                """
+            )
+            cur.execute("CREATE INDEX IF NOT EXISTS idx_servers_hostname ON servers(hostname);")
+        conn.commit()
+
+
+def row_to_dict(row: Iterable) -> dict:
+    id_, hostname, ip_address, state, created_at, updated_at = row
+    return {
+        "id": id_,
+        "hostname": hostname,
+        "ip_address": ip_address,
+        "state": state,
+        "created_at": created_at,
+        "updated_at": updated_at,
+    }
+
+
+__all__ = [
+    "ALLOWED_STATES",
+    "DEFAULT_DATABASE_URL",
+    "UniqueViolation",
+    "Connection",
+    "ConnectionPool",
+    "create_pool",
+    "ensure_schema",
+    "get_database_url",
+    "row_to_dict",
+]

app/main.py

@@ -0,0 +1,132 @@
+import logging
+from contextlib import asynccontextmanager
+from typing import Generator, List, Optional
+
+from fastapi import Depends, FastAPI, HTTPException, Response, status
+
+from app import schemas
+from app.db import (
+    Connection,
+    ConnectionPool,
+    UniqueViolation,
+    create_pool,
+    ensure_schema,
+    row_to_dict,
+)
+
+LOGGER = logging.getLogger(__name__)
+
+
+def create_app(database_url: Optional[str] = None) -> FastAPI:
+    pool: ConnectionPool = create_pool(database_url)
+
+    @asynccontextmanager
+    async def lifespan(app: FastAPI):
+        ensure_schema(pool)
+        yield
+        pool.close()
+
+    app = FastAPI(title="Server Inventory API", version="1.0.0", lifespan=lifespan)
+    app.state.pool = pool
+
+    def get_connection() -> Generator[Connection, None, None]:
+        with pool.connection() as conn:
+            yield conn
+
+    @app.get("/health", status_code=status.HTTP_200_OK)
+    def health() -> dict:
+        return {"status": "ok"}
+
+    @app.post("/servers", response_model=schemas.Server, status_code=status.HTTP_201_CREATED)
+    def create_server(server: schemas.ServerCreate, conn: Connection = Depends(get_connection)) -> dict:
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    """
+                    INSERT INTO servers (hostname, ip_address, state)
+                    VALUES (%s, %s, %s)
+                    RETURNING id, hostname, ip_address, state, created_at, updated_at
+                    """,
+                    (server.hostname, str(server.ip_address), server.state),
+                )
+                row = cur.fetchone()
+            conn.commit()
+            return row_to_dict(row)
+        except UniqueViolation:
+            conn.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail="A server with this hostname already exists."
+            )
+
+    @app.get("/servers", response_model=List[schemas.Server])
+    def list_servers(conn: Connection = Depends(get_connection)) -> list[dict]:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT id, hostname, ip_address, state, created_at, updated_at
+                FROM servers
+                ORDER BY id
+                """
+            )
+            rows = cur.fetchall()
+        return [row_to_dict(row) for row in rows]
+
+    @app.get("/servers/{server_id}", response_model=schemas.Server)
+    def get_server(server_id: int, conn: Connection = Depends(get_connection)) -> dict:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT id, hostname, ip_address, state, created_at, updated_at
+                FROM servers
+                WHERE id = %s
+                """,
+                (server_id,),
+            )
+            row = cur.fetchone()
+        if not row:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Server not found.")
+        return row_to_dict(row)
+
+    @app.put("/servers/{server_id}", response_model=schemas.Server)
+    def update_server(server_id: int, server: schemas.ServerUpdate, conn: Connection = Depends(get_connection)) -> dict:
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    """
+                    UPDATE servers
+                    SET hostname = %s,
+                        ip_address = %s,
+                        state = %s,
+                        updated_at = NOW()
+                    WHERE id = %s
+                    RETURNING id, hostname, ip_address, state, created_at, updated_at
+                    """,
+                    (server.hostname, str(server.ip_address), server.state, server_id),
+                )
+                row = cur.fetchone()
+            if not row:
+                conn.rollback()
+                raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Server not found.")
+            conn.commit()
+            return row_to_dict(row)
+        except UniqueViolation:
+            conn.rollback()
+            raise HTTPException(
+                status_code=status.HTTP_409_CONFLICT, detail="A server with this hostname already exists."
+            )
+
+    @app.delete("/servers/{server_id}", status_code=status.HTTP_204_NO_CONTENT)
+    def delete_server(server_id: int, conn: Connection = Depends(get_connection)) -> Response:
+        with conn.cursor() as cur:
+            cur.execute("DELETE FROM servers WHERE id = %s RETURNING id;", (server_id,))
+            row = cur.fetchone()
+        if not row:
+            conn.rollback()
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Server not found.")
+        conn.commit()
+        return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+    return app
+
+
+app = create_app()

app/schemas.py

@@ -0,0 +1,28 @@
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field, IPvAnyAddress, constr
+
+StateLiteral = Literal["active", "offline", "retired"]
+
+
+class ServerBase(BaseModel):
+    hostname: constr(strip_whitespace=True, min_length=1) = Field(..., description="Unique hostname for the server")
+    ip_address: IPvAnyAddress = Field(..., description="IPv4 or IPv6 address")
+    state: StateLiteral = Field(..., description="Operational state of the server")
+
+
+class ServerCreate(ServerBase):
+    pass
+
+
+class ServerUpdate(ServerBase):
+    pass
+
+
+class Server(ServerBase):
+    id: int
+    created_at: datetime
+    updated_at: datetime
+
+    model_config = ConfigDict(from_attributes=True)

cli/__init__.py

@@ -0,0 +1 @@
+"""Command-line utilities for managing the server inventory through the API."""