auth-service-fastapi

A production-oriented Authentication Microservice built with FastAPI, SQLAlchemy 2.0, and Redis.

Project Setup • Endpoints • System Design • Architecture • Token Security • Environment Variables • Future Improvements • Unit Tests • Integration Tests • E2E Tests • Idempotency

---

Project Setup

User Setup

docker compose -f docker-compose-dev.yml up -d

Developer Setup

1 — I'm using uv as package manager

curl -LsSf https://astral.sh/uv/install.sh | sh

2 — Install dependencies

uv sync

3 — Add your secrets to .env file

cp .env.example .env

4 — Start the infrastructure (PostgreSQL + Redis + pgAdmin)

docker compose -f src/docker-compose-dev.yml up -d

5 — Run the server

uv run python main.py

6 — Run the tests

uv run pytest

7 — API docs

http://localhost:8000/docs

8 — Run Tests

uv run pytest 

---

Endpoints

• /docs/ : interactive Swagger UI

Users

• /users/ :
  • POST — register a new user (creates user + profile atomically)
  • PUT — update user fields (name, family, username)
  • GET — list all users with filters and pagination
• /me/ :
  • GET — get the current authenticated user with profile
• /users-profile/ :
  • GET — list users joined with their profiles

Auth

• /auth/login :
  • POST — authenticate with username + password, returns access + refresh token pair
• /auth/refresh :
  • POST — issue a new access token using a valid refresh token
• /auth/logout :
  • POST — revoke both tokens (DB revocation + Redis blacklist)
• /auth/register-access :
  • POST — validate an access token and return its payload
• /auth/register-refresh :
  • POST — validate a refresh token and return payload + family_id

---

System Design

Request Lifecycle

Every authenticated request passes through two validation layers before reaching a route:

1. The access token JTI is checked against the Redis blacklist (fast O(1) lookup).
2. The token signature and claims (exp, iss, aud, nbf) are verified by python-jose.

This means even a structurally valid token is rejected instantly after logout, without touching PostgreSQL.

Session Lifecycle

Sessions are not injected via FastAPI Depends at the route level. Controllers open async with session_factory() explicitly and release the connection immediately after the DB step.

This prevents idle_in_transaction — a common production failure where connections stay open while the application does non-DB work (hashing, HTTP calls, cache writes), exhausting the connection pool under load.

Infrastructure

Client
  │
  ▼
FastAPI (Uvicorn)
  │
  ├── PostgreSQL 16 (asyncpg + SQLAlchemy 2.0)
  │     └── users, profiles, refresh_tokens
  │
  └── Redis 8 (async)
        └── access token blacklist
            refresh token blacklist (on logout)

Idempotency

Flow Breakdown

1. Atomic Lock Acquisition: The service attempts to write the key with a processing state using SET key "processing" NX EX ttl.
   • Success (Client 2): If the key does not exist, the operation returns True and the client proceeds with the operation.
   • Failure (Client 1): If the key already exists, Redis returns None/False.
2. Conflict Resolution: If lock acquisition fails, the service performs a GET to determine the current state of the request:
   • If state is processing, it raises IdempotencyException("Request already in progress. Please wait.").
   • If state is completed, it raises IdempotencyException("Duplicate request. Operation already completed.").
3. Execution Teardown: Once the business logic finishes successfully, the service calls mark_completed() to transition the state from processing to completed.

diagram

---

Architecture

This project follows a strict layered architecture. Dependencies only flow downward — routers know nothing about repositories, and repositories know nothing about JWT.

src/
├── main.py
├── auth/
│   ├── controller.py          — orchestrates login / refresh / logout
│   ├── routers.py
│   ├── repository.py
│   ├── dependencies.py
│   ├── schamas.py
│   └── expections.py
├── users/
│   ├── controllers.py         — orchestrates registration flow
│   ├── router.py
│   ├── service.py
│   ├── selectors.py           — all read queries
│   ├── repository.py
│   ├── dependencies.py
│   ├── schemas.py
│   └── exceptions.py
├── models/
│   ├── user.py                — User + Profile ORM models
│   ├── tokens.py              — RefreshTokenModel + ApiToken
│   └── enums.py
├── core/
│   ├── config.py              — settings via pydantic-settings
│   ├── base/
│   │   ├── repository.py      — Generic BaseRepository[T]
│   │   └── services.py        — BaseAbstractService (3-phase pattern)
│   └── security/
│       ├── jwt/               — TokenService + JWTController
│       ├── refresh_tokens/    — RefreshTokenStoreRepository
│       └── utils/             — hashing + token generation
├── infrastructure/
│   ├── sqlalchemy/            — AsyncSession, engine, declarative base
│   └── redis/                 — RedisManager + connection pool
└── test/
    ├── unit/
    ├── integration/
    ├── e2e/
    ├── fixtures/
    └── factories/

Three-phase Service Pattern

BaseAbstractService enforces a deliberate execution order to keep transactions tight:

• Pre-process — CPU-bound work outside the transaction (password hashing, token generation, input normalization)
• DB-process — only database writes inside async with session.begin(), no external awaits
• After-process — post-commit side effects (cache writes, event publishing, emails)

Key Patterns

• Repository Pattern — BaseRepository[T] is generic, providing find, create, update, delete, exists, and filter_by_* methods reused across all domain repositories.
• Dependency Injection — services, repositories, and infrastructure clients are injected at the controller level, keeping layers testable in isolation.
• Async-first — all DB and cache operations are fully async via asyncpg and aioredis.

---

Unit Tests

Unit tests target services only, not controllers.

Session is passed directly as a parameter to service methods, so tests run in complete isolation — no database, no Docker, no real connection needed. A lightweight Session_Mock() simulates all async session behaviour:

async def Session_Mock():
    session = AsyncMock()
    session.execute.side_effect  = execute   # prints query, sleeps 10ms
    session.commit.side_effect   = commit
    session.rollback.side_effect = rollback
    session.add.side_effect      = add
    session.flush.side_effect    = flush
    result = AsyncMock()
    result.scalar_one_or_none.return_value = None
    session.execute.return_value = result
    return session

UserRegistrationController is intentionally not unit tested — it contains no domain logic, only thin orchestration. It is covered by integration tests instead.

Mocking session_factory at the controller level would require reproducing async with session_factory() as session context manager behaviour — extra complexity with zero benefit.

---

Integration Tests

Integration tests cover controller → service → real DB round-trips using a real async_sessionmaker pointed at a test database (PostgreSQL running in Docker).

Covered flows:

• POST /users — user + profile created atomically, duplicate email/username raises correct error
• POST /auth/login — valid credentials return token pair, wrong password raises 401
• POST /auth/refresh — valid refresh token rotates and issues new access token
• POST /auth/logout — both tokens blacklisted, subsequent requests rejected

---

E2E Tests

End-to-end tests run the full HTTP stack via AsyncClient against a live application instance with real infrastructure (PostgreSQL + Redis).

Covered scenarios:

• Full registration → login → access protected route → logout → verify token rejected
• Refresh token reuse detection — second use of a spent refresh token revokes the entire family
• Concurrent login sessions — each session has an independent family_id

---

Token Security

Token Types

Token	TTL	Storage	Revocation
Access	ACCESS_TOKEN_EXPIRE_MINUTES (default 900)	Client-side only	Redis blacklist on logout
Refresh	REFRESH_TOKEN_EXPIRE_DAYS (default 43200)	PostgreSQL refresh_tokens table	DB revoked_at + Redis blacklist

Token Claims

{
  "sub":  "user_id",
  "jti":  "unique-token-id",
  "type": "access | refresh",
  "iat":  1234567890,
  "nbf":  1234567890,
  "exp":  1234567890,
  "iss":  "configured-issuer",
  "aud":  "Auth_service"
}

Token Family and Reuse Detection

Each login creates a family_id derived from the user-agent and IP hash. All refresh tokens in a session share this family_id. The RefreshTokenStoreRepository exposes revoke_family() to invalidate an entire session on detected reuse.

The rotated_from field on each refresh token stores the jti of the token it replaced, giving a full rotation lineage for audit.

Logout Flow

1. Verify both tokens are structurally valid.
2. Mark the refresh token as used and revoked in the DB (single transaction).
3. Blacklist the refresh token jti in Redis with TTL = remaining token lifetime.
4. Blacklist the access token jti in Redis with TTL = 5 minutes.

---

Environment Variables

Variable	Default	Description
DATABASE_URL	required	PostgreSQL DSN — asyncpg driver enforced automatically
DATABASE_URL_TEST	required	PostgreSQL DSN — asyncpg driver enforced automatically
REDIS_URL	required	Redis connection URL
REDIS_URL_TEST	required	Redis connection URL
SECRET_KEY	required	JWT signing secret
JWT_ALGORITHM	required	e.g. HS256
ISSUER	required	JWT iss claim value
AUDIENCE	Auth_service	JWT aud claim value
ACCESS_TOKEN_EXPIRE_MINUTES	900	Access token TTL in minutes
REFRESH_TOKEN_EXPIRE_DAYS	43200	Refresh token TTL in days
ENVIRONMENT	required	development or production
HOST	localhost	Uvicorn bind host
POST	8000	Uvicorn bind port
SQL_ECHO	False	Log all SQL statements (dev only)
IDEMPOTENCY_HEADER_NAME	X-Idempotency-Key	Idempotency-Header for each Request
IDEMPOTENCY_TTL	30	Idempotency time to live

---

Libraries

• FastAPI — async web framework
• SQLAlchemy 2.0 — async ORM with asyncpg driver
• python-jose — JWT signing and verification
• pydantic-settings — environment config
• Redis (aioredis) — token blacklist
• pytest — testing framework
• uv — package manager

---

Future Improvements

1. Wire /me route — the current-user JWT dependency exists but is not yet connected; cur_user is hardcoded to 1.

2. Tests — the structure is fully scaffolded (unit/, integration/, e2e/, fixtures/, factories/) but all files are empty. Integration tests for login, refresh, and logout are the critical path.

3. Email verification flow — the token_hash field and hashing utilities are in place; the verification endpoint and email dispatch are not yet implemented.

4. Redis caching for users — post-registration cache population is noted as a TODO in UserRegistrationController.

5. Alembic migrations — schema is currently created via Base.metadata.create_all; proper migration management is needed before production deployment.

6. aio-Logging debug async statements are scattered through service and controller layers and need to be replaced with structured logging sync.

7. Bulk user creation — endpoint code is scaffolded but commented out.

8. Bulk Update creation — endpoint code is scaffolded but commented out.

9. **Stremming Response ** — for Scroll or RealTime User Retreive