LogiTrack - Inventory & Logistics ERP API

An async FastAPI backend for managing inventory across multiple warehouses. The core of it is a stock-movement service that processes incoming, outgoing, and inter-warehouse transfer operations against a schema with real constraints enforced at the database level, plus role-based access control, JWT authentication with token revocation, idempotent stock movements, and a consistent Router -> Service -> Repository split across every domain.

Tech Stack

Layer	Technology
Language / Runtime	Python 3.12
API framework	FastAPI, Uvicorn
Database	PostgreSQL 16, SQLAlchemy 2.0 (async), Alembic
Auth	JWT (PyJWT), Argon2 password hashing (pwdlib)
Testing	pytest, pytest-asyncio, httpx (ASGI transport), PostgreSQL-backed integration fixtures
CI	GitHub Actions (Ruff lint & format, Mypy, pytest)
Linting	Ruff
Infra	Docker, Docker Compose

Architecture & Key Design Decisions

Layering

Every domain - product, category, supplier, warehouse, stock, user - follows the same three layers. Router handles HTTP and validation, Service holds the business rules, Repository does the data access. No SQL shows up in a service, no business logic shows up in a router. That split isn't just for tidiness, it's why test_stock_service.py can test StockService against a mocked repository with no database involved at all.

Data model

Product and Supplier connect through a product_supplier join table. Stock is limited to exactly one row per (product_id, warehouse_id) pair via a unique constraint, and quantity can never go negative, that's a CHECK (quantity >= 0) at the database level, not something left to application code to enforce. Every change to stock also writes an immutable StockMovement row, typed IN, OUT, or TRANSFER, with its own CHECK (quantity > 0).

flowchart TB

    %% =========================
    %% Catalog
    %% =========================
    subgraph Catalog["Catalog"]
        Category["Category"]
        Product["Product"]
        Supplier["Supplier"]
        ProductSupplier["product_supplier"]

        Category -->|"1 : N"| Product
        Product -->|"N : N"| ProductSupplier
        Supplier -->|"N : N"| ProductSupplier
    end

    %% =========================
    %% Inventory
    %% =========================
    subgraph Inventory["Inventory"]
        Warehouse["Warehouse"]
        Stock["Stock"]

        Product -->|"1 : N"| Stock
        Warehouse -->|"1 : N"| Stock
    end

    %% =========================
    %% Operations
    %% =========================
    subgraph Operations["Stock operations"]
        User["User"]
        StockMovement["StockMovement"]

        Product -->|"1 : N"| StockMovement
        User -->|"1 : N"| StockMovement

        Warehouse -->|"from_warehouse"| StockMovement
        Warehouse -->|"to_warehouse"| StockMovement
    end

    %% =========================
    %% Security
    %% =========================
    subgraph Security["Authentication support"]
        IdempotencyKey["IdempotencyKey"]
        RevokedToken["RevokedToken"]
    end

    %% =========================
    %% Visual emphasis
    %% =========================
    classDef main stroke-width:3px

    class Product,Stock,Warehouse,StockMovement main

Loading

Stock movement requests also support an optional Idempotency-Key. The server stores a request fingerprint and the resulting response so that retries with the same key and payload return the original result instead of applying the movement twice. Reusing an idempotency key with a different request payload results in a 409 Conflict.

The core logic: processing a stock movement

StockService.process_movement does three things, in order:

1. Checks the movement type against which warehouse fields are populated: IN needs only to_warehouse_id, OUT needs only from_warehouse_id, TRANSFER needs both, and they have to be different warehouses.
2. For OUT and TRANSFER, checks there's enough stock before debiting anything, raising a domain-specific InsufficientStockError if there isn't.
3. Updates the relevant Stock row(s) and inserts the audit record - all inside a single AsyncSession.

Atomicity comes from the session-per-request pattern: get_async_db commits the pending transaction at the end of the request and rolls back on any exception, so a TRANSFER that fails writing the destination side rolls back the source side with it - there's no state where only half a transfer went through. The only mid-request commit is the idempotency-key reservation, which is intentionally persisted before the movement runs so that a concurrent duplicate request can detect it.

The check-then-mutate sequence uses a row lock (SELECT ... FOR UPDATE) on the affected Stock row(s). When two concurrent requests hit the same (product_id, warehouse_id) pair, the database serializes them, forcing the second transaction to wait until the first commits or rolls back, so the check-then-debit sequence can never interleave. This complements the database-level CHECK (quantity >= 0) constraint.

The same lock protects the very first movement for a product-warehouse pair: when no Stock row exists yet, get_or_create_stock attempts to insert one inside a nested transaction, and if a concurrent request already created it, the unique constraint violation is caught and the winner's row is returned instead - a duplicate row can never slip through.

Idempotency

POST /stock/movements accepts an optional Idempotency-Key header. Each key is stored together with a SHA-256 request fingerprint, the authenticated user ID, the response status, and the serialized response body.

The key is reserved, and committed, before the movement is processed, so concurrent retries with the same key can never both run the business logic. A reservation starts in a pending state: a successful movement atomically completes it with the stored 201 response, while a failed movement is rolled back with its reservation cleaned up, leaving the key free for a later retry.

A repeated request with the same key and payload returns the previously stored response. If the same key is reused with a different payload, by a different user, or while another request with it is still in flight, the service raises ResourceConflictError, which is mapped to 409 Conflict.

Authentication & token revocation

Access tokens last 15 minutes and refresh tokens 7 days, with passwords hashed using Argon2. Both token types contain a unique jti identifier.

Logout revokes both the access and refresh token by storing their jti values in PostgreSQL. Every authenticated request checks whether its access token has been revoked, and refresh requests perform the same check for refresh tokens.

This makes logout effective server-side instead of relying only on client-side token removal.

Turning exceptions into HTTP responses

A small exception hierarchy - LogiTrackError, InsufficientStockError, InvalidMovementError, AlreadyExistsError, and ResourceConflictError - gets raised entirely inside the service layer and mapped to 400/409/500 responses by centralized FastAPI exception handlers. Services never import anything HTTP-related, which is exactly what makes them testable without the web layer in the picture.

Roles

Three roles - admin, warehouse_manager, operator - are enforced through a reusable RoleChecker dependency at the router level (require_auth, require_manager, require_admin). User management is admin-only; writing to catalog or warehouse data needs at least warehouse_manager.

Project Structure

LogiTrack/
├── alembic/                   # DB migrations
├── app/
│   ├── models/                # Product, Category, Supplier, Warehouse, Stock, StockMovement, User
│   ├── repositories/          # data access layer, one per aggregate
│   ├── services/              # business logic (stock transactions, RBAC, auth, idempotency)
│   ├── routers/               # REST endpoints, role-gated via RoleChecker
│   ├── schemas/               # Pydantic request/response models
│   ├── core/                  # db session, exceptions, security
│   ├── config.py
│   └── main.py
├── tests/{unit,integration}/
├── docker-compose.yaml        # postgres + api
└── Dockerfile

Getting Started

Docker (recommended)

git clone https://github.com/sa111nt/LogiTrack.git
cd LogiTrack
cp .env.example .env
# edit .env: set a real JWT_SECRET_KEY
docker compose up --build

The API is available at http://localhost:8000. Interactive docs at http://localhost:8000/docs (Swagger UI) or http://localhost:8000/redoc. Migrations run automatically on container start (alembic upgrade head).

Local (without Docker)

Requires a running PostgreSQL instance reachable via DATABASE_URL.

pip install -r requirements.txt
cp .env.example .env   # edit DATABASE_URL to point at your local Postgres
alembic upgrade head
uvicorn app.main:app --reload

Environment Variables

Variable	Required	Default	Description
DATABASE_URL	yes	-	Async PostgreSQL DSN (postgresql+asyncpg://...)
JWT_SECRET_KEY	yes	-	Sign this with a real secret, not the example value
JWT_ALGORITHM	no	HS256
ACCESS_TOKEN_EXPIRE_MINUTES	no	15
REFRESH_TOKEN_EXPIRE_DAYS	no	7
DEBUG	no	false	Enables SQL echo and verbose logging

API Overview

Full interactive documentation is generated automatically by FastAPI and served at /docs once the app is running - that's the source of truth for request/response schemas. Primary endpoints:

Method	Path	Min. role	Description
GET	/health	-	Liveness/version check
POST	/api/v1/auth/register	-	Register a new user
POST	/api/v1/auth/login	-	OAuth2 password flow, returns access + refresh tokens
POST	/api/v1/auth/refresh	-	Exchange a refresh token for a new pair
POST	/api/v1/auth/logout	any	Revoke the current access and refresh tokens
GET	/api/v1/auth/me	any	Current authenticated profile
*	/api/v1/users/*	admin	User management (CRUD)
*	/api/v1/categories/*, /suppliers/*	manager (write)	Catalog reference data
*	/api/v1/products/*	manager (write), any (read)	Product catalog
*	/api/v1/warehouses/*	manager (write), any (read)	Warehouse registry
POST	/api/v1/stock/movements	any	Process an IN / OUT / TRANSFER movement, with optional Idempotency-Key
GET	/api/v1/stock/movements	any	Movement history, filterable by type
GET	/api/v1/stock/warehouse/{id}	any	Current inventory at a warehouse
GET	/api/v1/stock/product/{id}	any	Stock levels for a product across all warehouses

Testing

pytest

Unit tests hit the JWT/password module and StockService's business rules directly against a mocked repository - movement type/warehouse validation and insufficient-stock handling - with no database involved.

Integration tests run through httpx's ASGI transport against a real PostgreSQL database. tests/conftest.py creates a dedicated logitrack_test database on the fly, rebuilds the schema for every test, and hands each request its own session with commit/rollback semantics matching production. That heavyweight setup exists for a reason: the concurrency guarantees - row locking, ordered stock writes, and idempotency reservation - depend on real Postgres behavior (SELECT ... FOR UPDATE, unique-constraint races), so a relational database is required to run the suite.

The integration coverage includes user registration and login, logout and token revocation, product and category routes, stock movements (IN/OUT/TRANSFER), idempotent retries returning the cached response, conflicting idempotency keys being rejected with 409, concurrent OUT movements being serialized so stock never goes negative, concurrent first stock creation producing a single Stock row, and concurrent requests with the same idempotency key running the business logic exactly once.

Continuous Integration

CI runs on every push and pull request to main via GitHub Actions (.github/workflows/ci.yml). The pipeline starts a PostgreSQL 16 service container, then runs, in order:

1. Ruff lint check (ruff check .)
2. Ruff format check (ruff format --check .)
3. Mypy type checking (mypy .)
4. The full test suite against the Postgres-backed fixtures (pytest -v)

A green pipeline means the code is linted, formatted, type-checked, and every concurrency/idempotency guarantee is verified against a real database.

License

This project is licensed under the MIT License. See the LICENSE file for details.