Just like we did with the Laravel microservice system, featuring two simple services: one for tasks and another for users. The joke writes itself: a task CRUD —we'll see it in a moment, we already have it here— and, to perform each operation, the user obviously has to be authenticated. And here the authentication is not local to this tasks application —it is simply a FastAPI application—: it depends on another service.
"Microservice", "service", and "application" will be used as synonyms.
In this case, the other service would be the user-service, and it connects to it via HTTP requests. Yes, there are more efficient ways —throughout this guide we will be looking at other things and how to configure them correctly—, but the idea is simply to give an introduction to microservices with FastAPI: the essence of FastAPI, mainly, and to go from the bare minimum to something a little more complete. So, once again, this is an introductory guide.
Finally, at the end of this post, you have access to the tagged source code so you can test section by section; this publication offers a presentation of the most important elements of microservice architecture.
First microservices, interconnected app
In this first part, we are going to use two microservices:
| service | port | what it does |
|---|---|---|
| user-service | 8000 | IdP: email/password login, opaque tokens with 60-min TTL |
| task-service | 8001 | User task CRUD; entrusts identity to the IdP via HTTP |
It is assumed that you have at least basic knowledge of FastAPI. If you don't: in my blog and academy platform there are many resources, from posts, community books that you can start completely for free just like the course, which I have both on my platform and on Udemy. The necessary knowledge is, more than anything, that you know how to make a basic CRUD and what it entails. So, now yes, let's begin.
No Server UI: Automatic Documentation
As mentioned before: we have two services. Unlike the Laravel application, which we were rendering with Blade, here I didn't want to do that in order to avoid adding views and complicating things a bit more, since I believe we can make better use of it later by connecting an application —for example, in Vue or React— and consuming the same services. Therefore, for now we are going to use automatic documentation, something we don't have in Laravel. And here it's pretty much the same: depending on the ecosystem you are in when we start developing, we can use one scheme or another. In this case, since we have automatic documentation here, it works wonderfully for doing these initial tests and connections, and then we see about spinning up a UI depending on the project's requirements, the client, or whatever. As I tell you, in Laravel usually —since it's a slightly more traditional framework— I already threw in Blade so as not to complicate things with Vue and so on.
Synchronous Communication Between Microservices for Authentication
In the early stages of designing a microservices architecture, identity validation and the processing of dependent operations are usually resolved through direct HTTP calls between the different services that make up the system.
1. Port and Endpoint Configuration
To allow microservices to interact with each other in a local development environment, it is necessary to define specific and stable ports. For example:
- User Service: Configured to operate on port
8000. - Task Service: Configured to operate on port
8001.
Since the task microservice needs to verify the user's permissions and identity before processing any request, the application makes direct HTTP requests to the port assigned to the user service.
2. Synchronous Authentication Flow
Under this scheme, every time the user performs an action in the task service, the latter issues a synchronous request to the authentication service to check the validity of the credentials or the access token.
Although this approach facilitates initial integration and ensures that validation is centralized in the user microservice, it introduces a strict runtime dependency: if the authentication service suffers high latency or goes down, dependent services will be directly affected.
The use of direct HTTP calls between microservices (such as those made from port
8001to8000) is a valid pattern for structuring basic authentication communication. However, as the system scales, these synchronous interactions are usually complemented or replaced by mechanisms based on self-signed tokens (such as JWT) or asynchronous message queues to improve the overall resilience of the architecture.
Notice that from the user service
task-service/app/config.py
import os
def _int(name: str, default: int) -> int:
return int(os.environ.get(name, str(default)))
# Where the IdP lives. In production, environment variable; in dev, port 8000.
USER_SERVICE_URL = os.environ.get("USER_SERVICE_URL", "http://127.0.0.1:8000")
HTTP_TIMEOUT_SECONDS = _int("HTTP_TIMEOUT_SECONDS", 2)
# sqlite for this service; in tests, DATABASE_URL=sqlite:///:memory:
DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///database.sqlite")HTTP Client Configuration: The Timeout
Notice that we have what was discussed: the user service with port 8000, and some additional configurations —as is totally normal—: the timeout. If it exceeds two seconds, we discard it, just in case there are any issues: a very basic protection that we need to improve little by little.
That timeout is what the HTTP client uses when making the request (we see it in block 14); and the IdP URL can be changed with the USER_SERVICE_URL environment variable, so the hardcoded 8000 is only the development value.
The Database (SQLite, SessionLocal, get_db)
We are using SQLite. Here we have the database connection, pretty much the standard: here we declare it —and this would be for tests, with sqlite:///:memory:—. Here we have the usual: the engine, SessionLocal, and get_db —the yield— to return the connection. Therefore, this is the method we are going to use via a dependency.
task-service/app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app import config
class Base(DeclarativeBase):
pass
def _make_engine(url: str) -> "create_engine": # type: ignore[valid-type]
kwargs = {"connect_args": {"check_same_thread": False}}
if url.endswith("//:memory:"):
from sqlalchemy.pool import StaticPool
kwargs["poolclass"] = StaticPool
return create_engine(url, **kwargs)
engine = _make_engine(config.DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()Starting the Application (main.py)
To start the application. Very normal: it's simply the lifespan —tables are created with it—, FastAPI, we boot up the application, and include_router, with the router from before.
task-service/app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db import Base, engine
from app.routes_tasks import router as tasks_router
@asynccontextmanager
async def lifespan(app: FastAPI):
Base.metadata.create_all(engine)
yield
app = FastAPI(
title="task-service",
description="Tasks per user, authenticated via the IdP (chapter 1).",
version="0.1.0",
lifespan=lifespan,
)
app.include_router(tasks_router)The Task Model: The User is an Integer, Not an FK
The task model is very simple: it's simply a base model and whatever you want to place: the id, the user —notice it's just an integer—, because here, once again, it doesn't know what on earth a user is. Therefore, we cannot have an FK-type relationship here, because, once again, by convention, the user lives in another application, their database also lives in that application, and absolutely nothing is known about it here. Otherwise, very normal: title, status, and whatever else you want to put in there.
task-service/app/models.py
from datetime import UTC, datetime
from sqlalchemy import DateTime, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from app.db import Base
STATUS_PENDING = "pending"
STATUS_COMPLETED = "completed"
STATUSES = (STATUS_PENDING, STATUS_COMPLETED)
def utcnow() -> datetime:
# Naive in UTC: sqlite does not preserve tzinfo.
return datetime.now(UTC).replace(tzinfo=None)
class Task(Base):
__tablename__ = "tasks"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(index=True)
title: Mapped[str] = mapped_column(Text)
status: Mapped[str] = mapped_column(String(20), default=STATUS_PENDING)
created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
updated_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow, onupdate=utcnow)The Schemas (Pydantic)
The schema: nothing weird. Just, well, the Pydantic ones as usual; and well, it would be TaskUpdate and TaskCreate, the ones we are going to handle; the output one —to render it on screen—, very normal; and the client one. normalize_status is the status validation: outside of pending|completed, a 422.
task-service/app/schemas.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, Field
from app.models import STATUSES
class TaskCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
status: str | None = None
class TaskUpdate(BaseModel):
title: str | None = Field(default=None, min_length=1, max_length=200)
status: str | None = None
class TaskOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
user_id: int
title: str
status: str
created_at: datetime
updated_at: datetime
def normalize_status(value: str | None, default: str) -> str:
if value is None:
return default
if value not in STATUSES:
from fastapi import HTTPException
raise HTTPException(
status_code=422,
detail=f"status debe ser uno de: {', '.join(STATUSES)}",
)
return valueTask CRUD and Inter-service Connection
Let's leave the routes one here, where everything connects. Obviously, because we have, for example, the task_id, to find a task, and that obviously I am the owner, so as not to see other users' tasks; all of this is done through here. With the task_id we search; if the task exists, we return a 200; if it is null, then go fly a kite: a 404; and if you are not the owner, well it's not your task: a 403.
And around here, as an interesting thing, we have current_user, which we will see what it is. Otherwise, it's the same flow for the rest of the endpoints: to create, simply, once again, the identified user —we always have this dependency, we'll see it shortly—; we create the task, insert it into the database. As a dependency here we also have get_db —you can see it right there, you have it as a dependency—. Just like getting all tasks: it is simply a query to our database, with the corresponding query, to get all the user's tasks, obviously with the current_user, you can see it here. Just like getting a particular task, or updating it, or deleting it: it's exactly the same thing we see in the basic course or book, so I won't go into much more detail.
task-service/app/routes_tasks.py
def find_owned(task_id: int, current_user: dict[str, Any], db: Session) -> Task:
task = db.get(Task, task_id)
if task is None:
raise HTTPException(status_code=404, detail="Tarea no encontrada")
if task.user_id != current_user["id"]:
# The 403 goes before any internal error: it doesn't exist for you.
raise HTTPException(status_code=403, detail="No es tu tarea")
return task
@router.post("/tasks", response_model=TaskOut, status_code=201)
def create_task(
payload: TaskCreate,
current_user: dict[str, Any] = Depends(get_current_user),
db: Session = Depends(get_db),
) -> Task:
task = Task(
title=payload.title,
status=normalize_status(payload.status, STATUS_PENDING),
# The owner is set by the system: the client does not send user_id.
user_id=current_user["id"],
)
db.add(task)
db.commit()
db.refresh(task)
return task
@router.get("/tasks", response_model=list[TaskOut])
def list_tasks(
current_user: dict[str, Any] = Depends(get_current_user),
db: Session = Depends(get_db),
) -> list[Task]:
return list(
db.query(Task)
.filter(Task.user_id == current_user["id"])
.order_by(Task.id.desc())
.all()
)(show_task, update_task, and delete_task have the same shape: always Depends(get_current_user) + Depends(get_db), and find_owned upfront.)
@router.get("/tasks/{task_id}", response_model=TaskOut)
def show_task(
task_id: int,
current_user: dict[str, Any] = Depends(get_current_user),
db: Session = Depends(get_db),
) -> Task:
return find_owned(task_id, current_user, db)Credential Management and Authentication in the Task Service (deps.py)
In the development of microservices-based architectures, validating the user's identity in secondary services (such as the task service) requires extracting and verifying credentials sent from the client or emitting service.
task-service/app/deps.py
from typing import Any
from fastapi import Header, HTTPException
from app.user_service_client import verify_token
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import Any
# 1. Define FastAPI's official security scheme
security = HTTPBearer()
def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict[str, Any]:
# credentials.scheme will be "Bearer" and credentials.credentials will be the clean token
token = credentials.credentials
print(token) # To verify that it arrives properly
user_data = verify_token(token)
if not user_data:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token no presentado o malformado"
)
return user_dataThat is what the previous method is for, which allows communicating with the user service which for now is simply support and this is what differentiates a traditional monolithic system from a microservice, the connection between them.
For now with a traditional HTTP connection with the main disadvantage of latency, but we will solve this in another section.
HTTP Client for Inter-Service Communication (user_service_client.py)
In a microservices architecture, communication between decoupled applications requires isolating network logic into specific modules. In this scheme, the task service needs to synchronously query the user service to validate identity via a token.
All external HTTP communication logic is concentrated in a single point (user_service_client.py), allowing the rest of the application modules to operate with user data as if they were local resources.
def verify_token(token: str) -> dict[str, Any]:
"""Who is this token? All network boundary lives in this file:
other modules send you the user dict as if it were local."""
try:
resp = _http_get(
f"{config.USER_SERVICE_URL}/api/user",
headers={"Authorization": f"Bearer {token}"},
)
except (httpx2.ConnectError, httpx2.ConnectTimeout):
# It's an infrastructure failure, not identity: the client deserves
# to know which service is missing and how to spin it up (503, not generic `401`).
raise HTTPException(
status_code=503,
detail=(
"user-service no está en pie (se esperaba "
f"{config.USER_SERVICE_URL}); levántalo con "
"`uv run uvicorn app.main:app --port 8000` en user-service/"
),
)
if resp.status_code == 200:
user = resp.json()
# The IdP client returns {id, name, email}; nothing else enters here.
return {"id": int(user["id"]), "name": str(user["name"]), "email": str(user["email"])}
if resp.status_code == 401:
# The distinction that matters: `401` is NOT a service failure, it is
# invalid identity of the presented token.
raise HTTPException(status_code=401, detail="Token inválido o expirado")
raise HTTPException(
status_code=502,
detail=f"Respuesta inesperada del user-service (HTTP {resp.status_code})",
)
Error Handling and Network Resilience
By relying on synchronous HTTP calls between services, exception handling must precisely distinguish between infrastructure failures and authentication problems:
- Infrastructure failure (HTTP 503): If the user service is unavailable (catching connection or timeout errors in
httpx2), it is interpreted as an operational system outage rather than a credential error, instructing the operator on how to spin up the affected service. - Invalid identity (HTTP 401): If the service responds indicating that the token is invalid or has expired, an authentication error is propagated to the client.
- Unexpected responses (HTTP 502): Any other unhandled status code translates into a gateway failure due to anomalous responses from the identity provider.
Conclusion
Centralizing token validation through a dedicated HTTP client simplifies interoperability between microservices. Although this approach introduces latency per request compared to decentralized schemes (such as self-signed JWT tokens), it provides strict control over session state in these initial stages of development.
These are typical validations: if 200 is returned, then we have the user; and if not, invalid or expired token. And that's all we have.
task-service/app/user_service_client.py
def verify_token(token: str) -> dict[str, Any]:
"""Who is this token? All network boundary lives in this file:
other modules send you the user dict as if it were local."""
try:
resp = _http_get(
f"{config.USER_SERVICE_URL}/api/user",
headers={"Authorization": f"Bearer {token}"},
)
except (httpx2.ConnectError, httpx2.ConnectTimeout):
# It's an infrastructure failure, not identity: the client deserves
# to know which service is missing and how to spin it up (503, not generic `401`).
raise HTTPException(
status_code=503,
detail=(
"user-service no está en pie (se esperaba "
f"{config.USER_SERVICE_URL}); levántalo con "
"`uv run uvicorn app.main:app --port 8000` en user-service/"
),
)
if resp.status_code == 200:
user = resp.json()
# The IdP client returns {id, name, email}; nothing else enters here.
return {"id": int(user["id"]), "name": str(user["name"]), "email": str(user["email"])}
if resp.status_code == 401:
# The distinction that matters: `401` is NOT a service failure, it is
# invalid identity of the presented token.
raise HTTPException(status_code=401, detail="Token inválido o expirado")
raise HTTPException(
status_code=502,
detail=f"Respuesta inesperada del user-service (HTTP {resp.status_code})",
)(This is the client with the timeout from block 7: httpx2.Client(timeout=config.HTTP_TIMEOUT_SECONDS); if the request takes more than two seconds, it gets discarded.)
user-service: the seed and security (hash and token)
There are many ways to do this, so don't dwell on it too much. They are utility functions, obviously, to manipulate the token —this would be to compare the token: hash_token—; they are functions of that type. So I think it's understood a little bit there.
user-service/app/seed.py
"""First part demo users (same as Laravel series):
uv run python -m app.seed
"""
from sqlalchemy.orm import Session
from app import security
from app.db import Base, engine, SessionLocal
from app.models import User
DEMO_USERS = [
{"name": "Ana Torres", "email": "ana.torres@example.com", "password": "password"},
{"name": "Luis Ramirez", "email": "luis.ramirez@example.com", "password": "password"},
]
def run() -> None:
Base.metadata.create_all(engine)
db: Session = SessionLocal()
try:
for spec in DEMO_USERS:
exists = db.query(User).filter(User.email == spec["email"]).first()
if exists:
continue
db.add(
User(
name=spec["name"],
email=spec["email"],
password=security.hash_password(spec["password"]),
)
)
db.commit()
users = db.query(User).count()
print(f"{users} usuario(s) en la base de datos")
finally:
db.close()
if __name__ == "__main__":
run()user-service/app/security.py
import hashlib
import hmac
import secrets
from datetime import UTC, datetime, timedelta
def hash_password(password: str) -> str:
# Standard library PBKDF2: no dependencies, deterministic.
salt = secrets.token_hex(16)
digest = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100_000).hex()
return f"pbkdf2_sha256$100000${salt}${digest}"
def verify_password(password: str, stored: str) -> bool:
try:
_algo, _iters, salt, digest = stored.split("$")
except ValueError:
return False
candidate = hashlib.pbkdf2_hmac("sha256", password.encode(), salt.encode(), 100_000).hex()
# Constant-time comparison: never direct `==`.
return hmac.compare_digest(candidate, digest)
def issue_token() -> tuple[str, str]:
"""Returns (raw token, hash to store). The raw one travels only once."""
token = secrets.token_hex(32)
return token, hashlib.sha256(token.encode()).hexdigest()
def hash_token(token: str) -> str:
return hashlib.sha256(token.encode()).hexdigest()
def expiry(now: datetime | None = None) -> datetime:
from app import config
# Naive UTC for the column; the route makes it aware when responding.
return (now or models_now_naive()) + timedelta(minutes=config.TOKEN_TTL_MINUTES)The User Schemas
The schema we have, very normal, for the token and the user, are the Pydantic models.
user-service/app/schemas.py
from datetime import datetime
from pydantic import BaseModel, ConfigDict, EmailStr, Field
class TokenRequest(BaseModel):
email: EmailStr
password: str = Field(min_length=1)
device_name: str | None = Field(default="api-token", max_length=100)
class UserInfo(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
email: EmailStr
class TokenIssued(BaseModel):
token: str
token_type: str
expires_at: datetime
user: UserInfoThe User and Token Model (FK, hash, expiration)
And the model, typical: the user, and a model completely customized by us, without any standard or anything like that: the user, the token hash, the name, and expiration and creation dates. And the user_id, of course: in this case it is of relational type, to have the FK in the DB. Once again, here we have a completely independent database.
user-service/app/models.py
from datetime import UTC, datetime
from sqlalchemy import DateTime, ForeignKey, String
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db import Base
def utcnow() -> datetime:
# Naive in UTC: sqlite does not preserve tzinfo; the whole DB lives in UTC-naive.
return datetime.now(UTC).replace(tzinfo=None)
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
password: Mapped[str] = mapped_column(String(255))
created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
class ApiToken(Base):
__tablename__ = "api_tokens"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), index=True)
# Opaque token: only the hash is stored, the raw one leaves login once.
token_hash: Mapped[str] = mapped_column(String(64), unique=True, index=True)
name: Mapped[str] = mapped_column(String(100), default="api-token")
expires_at: Mapped[datetime] = mapped_column(DateTime)
created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow)
user: Mapped[User] = relationship(User)The DB, configuration, and user main
Here you have the database: the same connection as before; nothing changes from here. The configurations: pretty much the same —this would be the token duration, how long we want it to last, which would be one hour—. And in main we have, to boot up the application: models are created, routes...
user-service/app/db.py (identical in form to task-service's)
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
from app import config
class Base(DeclarativeBase):
pass
def _make_engine(url: str) -> "create_engine": # type: ignore[valid-type]
kwargs = {"connect_args": {"check_same_thread": False}}
if url.endswith("//:memory:"):
from sqlalchemy.pool import StaticPool
kwargs["poolclass"] = StaticPool
return create_engine(url, **kwargs)
engine = _make_engine(config.DATABASE_URL)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()The IdP Routes: login, /user, and revoke
The routes: getting the user; deleting the token —which would again be the database connection: getting the authorization token that we have to pass to it, deleting it, in this case we do it locally here—. And the login: we verify the user, get the plain text password, verify it against the hash in the database; if they match, perfect. And the token: the raw one is returned only once —only its hash arrives at the database—. Authenticate, which is what /api/user and DELETE use, looks up the token hash in the DB and checks that it has not expired.
user-service/app/routes_tokens.py
@router.post("/tokens", response_model=TokenIssued, status_code=201)
def issue_token(
payload: TokenRequest,
db: Session = Depends(get_db),
) -> TokenIssued:
"""Log in, once: email + password → opaque token that expires."""
user = db.query(User).filter(User.email == payload.email).first()
# Same message and same status for "does not exist" and "bad password": otherwise,
# this response can be used to enumerate which emails are registered.
if user is None or not security.verify_password(payload.password, user.password):
raise HTTPException(status_code=401, detail="Invalid credentials")
token, token_hash = security.issue_token()
expires_at = security.expiry()
row = ApiToken(
user_id=user.id,
token_hash=token_hash,
name=payload.device_name or "api-token",
expires_at=expires_at,
)
db.add(row)
db.commit()
return TokenIssued(
token=token,
token_type="Bearer",
expires_at=expires_at.replace(tzinfo=UTC),
user=UserInfo.model_validate(user),
)
@router.get("/user", response_model=UserInfo)
def whoami(user: User = Depends(require_user)) -> UserInfo:
return UserInfo.model_validate(user)
@router.delete("/tokens", status_code=204)
def revoke(
authorization: str | None = Header(default=None),
db: Session = Depends(get_db),
) -> None:
"""Revokes the token with which the request was made: deleting the row is a
real and immediate revocation (the opaque token trade-off)."""
from app.auth import authenticate
user, row = authenticate(db, authorization)
db.query(ApiToken).filter(ApiToken.id == row.id).delete()
db.commit()user-service/app/auth.py
def authenticate(db: Session, authorization: str | None) -> tuple[User, ApiToken]:
"""Who is this token? The response is `401` upon an invalid or expired token."""
token = _bearer_token(authorization)
if token is None:
raise HTTPException(status_code=401, detail="Token not presented or malformed")
row = db.query(ApiToken).filter(ApiToken.token_hash == security.hash_token(token)).first()
if row is None:
raise HTTPException(status_code=401, detail="Invalid or expired token")
# Expired token: for the client it is identical to a nonexistent one.
if row.expires_at <= utcnow():
raise HTTPException(status_code=401, detail="Invalid or expired token")
return row.user, rowConclusions and Microservices Definition
To conclude this block on distributed architecture, it is essential to understand both the conceptual definition of a microservice and the practical challenges that arise when interconnecting multiple applications synchronously.
1. Definition of a Microservice
A microservice is a software development architectural style in which a complete application (such as a task management system with identified users) is broken down into small, independent, and autonomous services. Each of these services fulfills a specific and well-bounded business function —in this case, separating user management from task processing—.
2. Communication and Initial Coupling
For the business logic to function in an integrated manner, microservices need to communicate with each other. In the initial stages, this interoperability is resolved through direct HTTP requests (for example, querying the user service to verify identity before executing an action in the task service).
However, this synchronous approach reveals the real challenges of distributed architecture versus theory:
- Operational dependency: If a critical service (such as authentication) stops responding, dependent services are immediately blocked, demonstrating that absolute autonomy requires additional resilience mechanisms.
- Management complexity: As the system scales, a large portion of development effort stops focusing on business logic and becomes dedicated to maintaining connections, the network, and the communication infrastructure between services.
Brute-Force Protection and Rate Limiting
In the design of distributed architectures and microservices, the implementation of Rate Limiting mechanisms is essential to protect critical access points against automated access, brute-force attacks, and resource saturation.
1. Philosophy and Objectives of Rate Limiting
The primary goal of traffic control is not to affect legitimate users, but to safeguard the application against malicious or uncontrolled requests. Just as server-side validations guarantee data input integrity (preventing unauthorized modifications to sensitive fields like user roles), Rate Limiting acts as a preventive barrier at the system's boundaries.
In microservice environments, where multiple applications communicate autonomously, authentication and access services are the most frequent targets of denial-of-service attacks or credential scraping, making their protection rigorous.
2. Dual Control Strategy: By IP and by Account
To effectively mitigate brute-force attacks targeting logins, the system implements a dual-counter strategy that evaluates two distinct threat vectors:
- IP address control: Stops massive request sweeps or automated access attempts originating from the same network address.
- User account control (combination of email and IP): Prevents a specific account from being systematically hammered.
It is important to note that the system counts all attempts (regardless of whether the outcome is successful or erroneous). A mechanism that limits only failures is insufficient, as it would allow continuous brute-force attacks until guessing valid credentials.
3. Technical Implementation in FastAPI
The control logic is encapsulated in a dedicated module that uses thread-safe in-memory dictionaries (threads) and monotonic timestamps to prevent system clock alterations:
user-service/app/rate_limit.py
import hashlib
import threading
import time
from collections import defaultdict, deque
from fastapi import HTTPException, Request
from app import config
_lock = threading.Lock()
_hits: dict[str, deque[float]] = defaultdict(deque)
def client_ip(request: Request) -> str:
return request.client.host if request.client else "unknown"
def _email_key(email: str) -> str:
return hashlib.sha256(email.strip().lower().encode()).hexdigest()
def check_login(request: Request, email: str) -> None:
now = time.monotonic()
ip = client_ip(request)
pairs = (
(f"ip:{ip}", config.LOGIN_THROTTLE_PER_IP),
(f"email:{_email_key(email)}|ip:{ip}", config.LOGIN_THROTTLE_PER_ACCOUNT),
)
with _lock:
for key, limit in pairs:
hits = _hits[key]
while hits and now - hits[0] >= config.LOGIN_THROTTLE_DECAY_SECONDS:
hits.popleft()
if len(hits) >= limit:
wait = int(config.LOGIN_THROTTLE_DECAY_SECONDS - (now - hits[0])) + 1
raise HTTPException(
status_code=429,
detail="Too many login attempts; try again later",
headers={"Retry-After": str(wait)},
)
for key, _limit in pairs:
_hits[key].append(now)
def reset() -> None:
with _lock:
_hits.clear()This mechanism integrates directly into routers responsible for processing credentials (such as the token generation flow), intercepting requests that exceed configured thresholds and returning an HTTP 429 (Too Many Requests) response accompanied by the Retry-After header.
user-service/app/routes_tokens.py
def rate_limit_login(request: Request, payload: TokenRequest) -> None:
"""Brute force against the endpoint receiving passwords: two counters
with the limits and window from app/config.py."""
check_login(request, payload.email)The incorporation of a robust rate limiting system at authentication endpoints significantly increases the perimeter security of microservices. By combining restrictions based on network origin and user identity, operational abuses are prevented without altering the browsing experience of legitimate users.
Observability and Traceability in Microservices Architecture
In interconnected systems, observability ceases to be a secondary feature and becomes an essential requirement. When a user request crosses multiple microservices (for example, linking the task service with the authentication service), any failure at an intermediate point is complex to isolate if a unified tracking mechanism is not available.
1. The Challenge of Tracking in Distributed Systems
At the interface level, the final client receives only a generic error code (such as a 500 failure). However, in architectures composed of diverse microservices operating in parallel, reconstructing the exact path of a request is practically impossible without the support of logging and traceability tools.
To solve this problem, a unique identifier per request is implemented, called X-Request-Id. If the incoming request does not possess this header, the system generates it automatically and propagates it throughout the entire call chain until the interaction ends.
2. The Traceability Contract
The observability mechanism is based on three key principles:
- Respect or create: If the client or a previous service sends the
X-Request-Idheader, the system reuses it; otherwise, it generates a new UUID (uuid.uuid4()). - Propagation in the response: The application always returns the
X-Request-Idheader in HTTP responses. - Automatic integration in logs: Any log message (
logger.info(...)) emitted during the request includes the identifier transparently, without requiring its manual passing between functions.
3. Technical Breakdown of Components
A. Context Variable (ContextVar)
request_id: ContextVar[str | None] = ContextVar("request_id", default=None)This asynchronous-programming-safe variable (in Python / asyncio) ensures that, in concurrent environments where thousands of requests execute simultaneously on the same thread, the identifier of a specific request never mixes with another.
B. Middleware (RequestIdMiddleware)
app/observability.py
class RequestIdMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
rid = request.headers.get(REQUEST_ID_HEADER) or str(uuid.uuid4())
token = request_id.set(rid)
try:
response = await call_next(request)
finally:
request_id.reset(token)
response.headers[REQUEST_ID_HEADER] = rid
return response- Upon entry: Evaluates whether the request includes the
X-Request-Idheader; otherwise, it creates a UUID and stores it in the context variable viarequest_id.set(rid). - During the request:
call_next(request)executes the endpoint and its dependencies, allowing any component to query the identifier withrequest_id.get(). - Upon exit (finally block): Clears the context variable (
request_id.reset(token)) to prevent contamination of future requests on the same thread. - In the response: Adds
response.headers[REQUEST_ID_HEADER] = ridto inform the client or emitting service of the code associated with its execution.
C. Log Filter (RequestIdFilter)
class RequestIdFilter(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
record.request_id = request_id.get() or "-"
return TrueEvery time a log statement executes, Python instantiates a LogRecord object. This filter intercepts said log before emission and dynamically injects the record.request_id attribute obtained from the context variable, guaranteeing its correct capture against testing tools.
D. Formatter (RequestIdFormatter)
class RequestIdFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
if not hasattr(record, "request_id"):
record.request_id = "-"
return super().format(record)Takes the request identifier (or a dash in case of events independent of the HTTP flow, such as server startup logs) and stamps it in a standardized way in the console output.
E. Configuration Function (configure_logging)
def configure_logging(*loggers: str) -> None:
logging.basicConfig(level=logging.INFO)
formatter = RequestIdFormatter(
"%(levelname)s:%(name)s:%(message)s request_id=%(request_id)s"
)
for handler in logging.getLogger().handlers:
handler.setFormatter(formatter)
for name in loggers:
logging.getLogger(name).addFilter(RequestIdFilter())Configures the base logging level to INFO, sets the visual message template, and associates both the formatter and the custom filter to the modules specified by parameter.
4. Propagation Between Independent Projects
Given that microservices operate as independent projects that do not share codebases through coupled libraries, each service must register this same traceability module.
When an emitting service makes an HTTP request to another microservice, it must extract its current identifier (request_id.get()) and send it explicitly in the outgoing X-Request-Id header. Upon receiving the request, the receiving service's middleware will detect the existing header, respect it, and unify all its logs under the same original identifier.
For example, when authenticated, if we place a log inside the task controller for listing:
task-service/app/routes_tasks.py
import json
import logging
log = logging.getLogger("task-service")
@router.get("/tasks", response_model=list[TaskOut])
def list_tasks(
current_user: dict[str, Any] = Depends(get_current_user),
db: Session = Depends(get_db),
) -> list[Task]:
log.info(
"Failed login attempt %s",
json.dumps(
{
"task": "ok",
},
ensure_ascii=False,
),
)
return list(
db.query(Task)
.filter(Task.user_id == current_user["id"])
.order_by(Task.id.desc())
.all()
)Let's remember that in each task request we send a request to the user service to check if we are authenticated to perform any CRUD operation from tasks, in this user service, we also place a log:
user-service/app/routes_tokens.py
import json
import logging
log = logging.getLogger("task-service")
@router.get("/user", response_model=UserInfo)
def whoami(user: User = Depends(require_user)) -> UserInfo:
log.info(
"API token issued %s",
json.dumps(
{
"user_id": user.id,
"name": user.name
},
ensure_ascii=False,
),
)
return UserInfo.model_validate(user)If we check the console (for now we have it set to print to the console, in production we should write them to a file), we will see the logs in both services and notice that it is the exact same ID, therefore, we can perform traceability between both microservices:
INFO:task-service:Failed login attempt {"task": "ok"} request_id=e9944cee-1215-4c86-86c6-f2a0f220c691INFO: 127.0.0.1:53451 - "GET /api/user HTTP/1.1" 200 OK
INFO:user-service:API token issued {"user_id": 1, "name": "Ana Torres"} request_id=e9944cee-1215-4c86-86c6-f2a0f220c691And this is thanks to this line in the Middleware:
rid = request.headers.get(REQUEST_ID_HEADER) or str(uuid.uuid4())Which allows generating or obtaining an identifier for the trace and propagating it in the response (which would be the request of the next call in the connection chain of our microservices):
response.headers[REQUEST_ID_HEADER] = rid