FastAPI Template

Purpose and Scope

This project serves as a template for developing FastAPI (Python) applications following the principles of Hexagonal Architecture (Ports and Adapters). The goal is to promote a decoupled, modular, and maintainable code structure, facilitating application evolution and testing.

Problem It Solves: It provides a structured foundation for building backend services in Python, clearly separating business logic from infrastructure concerns (databases, messaging, external APIs).

Scope:

• REST API implementation with FastAPI.
• Project structure based on Hexagonal Architecture (Domain, Application, Infrastructure).
• Optional integration with:
  • PostgreSQL database (with in-memory alternative).
  • Messaging with Apache Kafka.
  • Messaging with Google Cloud Pub/Sub.
• Configuration for local execution and deployment on GKE and Cloud Run.
• Unit test examples.
• Basic authentication mechanism.

Architecture

The template follows a Hexagonal Architecture (or Ports and Adapters), which organizes code into three main layers:

1. Domain (<modulo>/domain):

   • Contains pure domain models, core business logic, and interfaces (ports) for repositories and services.
   • Does not depend on any other layer.
   • Examples: app/main/user/domain/model/user.py, app/main/user/domain/repositories/user_repository.py (interface).

2. Application (<modulo>/application):

   • Orchestrates application use cases.
   • Implements service interfaces defined in the domain.
   • Uses ports (repository interfaces) to interact with infrastructure.
   • Contains DTOs (Data Transfer Objects) and application-specific logic.
   • Examples: app/main/user/application/user_service_impl.py, app/main/user/application/dtos/request_create_dto.py.

3. Infrastructure (<modulo>/infrastructure):

   • Contains concrete implementations of ports (adapters) and all logic related to external technologies.
   • Input Adapters: Expose the application to the outside world (e.g. API controllers, Kafka/PubSub consumers).
     • Examples: app/main/user/infrastructure/adapter/input/user_controller.py, app/main/user/infrastructure/adapter/input/kafka_receiver.py.
   • Output Adapters: Implement interaction with external services (e.g. database repositories, Kafka/PubSub publishers).
     • Examples: app/main/user/infrastructure/adapter/output/pg_user_storage.py, app/main/message/infrastructure/kafka.py.

Directory Structure (Example for user module):

main/
└── user/
    ├── application/
    │   ├── config/             # Module-specific configuration (e.g. DB connection)
    │   ├── dtos/               # Data Transfer Objects
    │   ├── entities/           # Persistence entities (if applicable)
    │   └── user_service_impl.py # Use case implementation
    ├── domain/
    │   ├── model/              # Domain models
    │   ├── repositories/       # Repository interfaces (ports)
    │   └── services/           # Service interfaces (ports)
    └── infrastructure/
        └── adapter/
            ├── input/
            │   ├── api/        # API controllers (FastAPI endpoints)
            │   └── kafka_receiver.py # Kafka consumer
            │   └── pubsub_receiver.py# PubSub consumer
            └── output/
                ├── pg_user_storage.py # PostgreSQL repository implementation
                └── in_memory_user_repository.py # In-memory repository implementation

Technologies and Dependencies

• Language: Python (>=3.11)
• API Framework: FastAPI
• ASGI Server: Uvicorn
• WSGI/ASGI Server (Production): Gunicorn
• Dependency Management: Pipenv, pip
• Database (Optional): PostgreSQL (with SQLAlchemy as ORM)
• Messaging (Optional):
  • Apache Kafka (with confluent-kafka)
  • Google Cloud Pub/Sub (with google-cloud-pubsub)
• Testing: Pytest, Coverage.py
• Authentication: Google OAuth2 Client (for token validation)
• Containerization: Docker
• CI/CD: Jenkins, GitLab CI
• Orchestration (Deployment): Google Kubernetes Engine (GKE), Google Cloud Run

List of main dependencies (see requirements.txt for the full list):

fastapi
uvicorn
gunicorn
pipenv
# For PostgreSQL
sqlalchemy
psycopg2-binary
# For Kafka
confluent-kafka
# For Pub/Sub
google-cloud-pubsub
# For Auth
google-auth
google-api-python-client
# For Testing
pytest
coverage

Local Setup

Prerequisites

• Python 3.11 or higher.
• pip installed.
• Recommended IDE: Visual Studio Code.
• (Optional) Docker and Docker Compose for dependencies such as PostgreSQL, Kafka.
• (Optional) Access to GCP services if using Pub/Sub or Kafka on Confluent Cloud and you want to test against them.

Setup Steps

1. Clone the repository: Follow these steps

2. Install Pipenv:

   pip install pipenv

3. Activate the virtual environment:

    pipenv shell

   Ensure your IDE (VS Code) uses this virtual environment. It is usually located at ~/.virtualenvs/template_fast_api-XXXXXX.

4. Install project dependencies:

   pipenv install -r requirements.txt

   This will install the dependencies listed in requirements.txt and create/update Pipfile and Pipfile.lock.

5. Create .env file: At the project root, create a .env file with the following basic structure. Uncomment and configure the sections according to the components you will use.

   ENVIRONMENT=dev
   DEBUG=true # or false
   
   # Database Configuration (PostgreSQL) - Optional
   DATABASE_USERNAME=postgres
   DATABASE_PASSWORD=postgres
   DATABASE_NAME=hexa
   DATABASE_HOST=localhost
   DATABASE_PORT="5432"
   SCHEMA_NAME=public
   
   # Kafka Configuration (Confluent Cloud or Local) - Optional
   # Confluent Cloud example:
   BOOTSTRAP_SERVER=pkc-xxxx.xxxx.gcp.confluent.cloud:9092
   CLUSTER_API_KEY=TU_KAFKA_API_KEY
   CLUSTER_API_SECRET=TU_KAFKA_API_SECRET
   # Local example:
   # BOOTSTRAP_SERVER=localhost:9092
   TOPIC_RECIEVER=tu_topic_kafka_consumidor # Topic the app consumes
   TOPIC_PRODUCER=tu_topic_kafka_productor # Topic the app produces to (configured in kafka.py)
   RETRIES=10
   CONSUMER_GROUP_ID=tu_grupo_consumidor_kafka
   
   # Google Cloud Pub/Sub Configuration - Optional
   # Used by PubsubConsumer
   PROJECT_ID=tu-gcp-project-id
   SUBSCRIPTION_ID=tu-pubsub-subscription-id-para-consumir
   # Used by PubsubNotifier (main.py) - Ensure they match or adjust as needed
   PUBSUB_PROJECT_ID=tu-gcp-project-id # Can be the same as PROJECT_ID
   PUBSUB_TOPIC_ID=tu-pubsub-topic-id-para-producir
   # GOOGLE_APPLICATION_CREDENTIALS=/ruta/a/tus/credenciales-gcp.json # Required if not using ADC
   
   # Authentication Configuration (Google OAuth2) - Optional
   GOOGLE_CLIENT_ID=tu-google-client-id.apps.googleusercontent.com
   
   # Logging Configuration
   LOGGING_SEVERITY="DEBUG"

   Note: The .env file is included in .gitignore and must not be versioned. For Kafka and Pub/Sub, if you do not have local instances, you can use emulators or cloud services.

6. Run the application locally:

   uvicorn main:app --reload --port 8080

   Or using the main script if configured for uvicorn.run:

   python main.py

   The application will be available at http://localhost:8080. Swagger UI documentation at http://localhost:8080/api/hexa/docs.

New Dependencies

To install new dependencies:

pipenv install <nombre_dependencia>

To update requirements.txt from Pipfile.lock (to maintain consistency with Dockerfile or other setups):

pipenv run pip freeze > requirements.txt

Deployment

The project is configured to deploy on Google Kubernetes Engine (GKE) or Google Cloud Run.

GKE Deployment

• Uses the Jenkinsfile with the gke-krane-python pipeline:

  pipelineLatam("gke-krane-python")

• Deployment configuration (Deployment, Service, etc.) is in deploy/gke/deployment.yaml.erb.
• Environment-specific variables (dev, intg, prod) are managed through ConfigMaps and Secrets in Kubernetes, populated from deploy/env/<ambiente>.yaml.

Cloud Run Deployment

• Uses .gitlab-ci.yml which includes the Cloud Run pipeline component:

  include:
    - component: $LATAM_PIPELINE_CLOUD_RUN_DEPLOY

• Cloud Run service configuration is in deploy/run/runtime-config.yaml.erb.
• Environment variables are defined in deploy/run/environment.yaml.erb and are based on deploy/env/<ambiente>.yaml files.

Common Environment Variables for Deployment

See files in deploy/env/ for environment-specific variables, such as:

• PROJECT_ID
• CLOUD_RUN_REGION
• Database credentials (usually injected as secrets)
• Kafka/PubSub configuration (endpoints, topics, credentials as secrets)
• GOOGLE_CLIENT_ID for authentication.

API Endpoints

The base API is at /api/hexa. All endpoints in the user module require a uuid header.

Method	Endpoint	Authentication	Description	Example Request Body (JSON)	Example Success Response
GET	/api/hexa/how-use-it	None	Returns link to template documentation	N/A	"https://docsrepo.appslatam.com/display/E2EDEV/Python+Template+FastAPI" (plain text)
POST	/api/hexa/user/registration	Header uuid	Registers a new user.	{"username": "newuser", "email": "new@example.com"}	{"status_code": 200, "data": "user created"}
POST	/api/hexa/user/get_user	Header uuid	Gets a user by username.	{"username": "testuser", "email": "ignored@example.com"}	{"status_code": 200, "data": "{'username': 'testuser', 'email': 'test@example.com'}"}

General Authentication (Optional): The system includes a Google OAuth2 token validator in app/main/auth/infrastructure/auth.py. If enabled on endpoints (using Depends(validate_token)), an Authorization: Bearer <token_id_google> header will be required. Currently, user_controller endpoints only depend on the uuid header.

Example Error Responses for /registration and /get_user:

• User already exists (/registration): {"status_code": 201, "data": "username already exists"}
• Validation failed (/registration): {"status_code": 202, "data": "username must not contain digits"}
• User not found (/get_user): {"status_code": 404, "data": "user not found"}
• Missing/invalid uuid header: 400 Bad Request with detail.

Specific Components

Database (PostgreSQL / In Memory)

• Description: The template allows using a PostgreSQL database or an in-memory implementation for persistence.

• Selection: In main.py, the repository to use is selected:

  # For PostgreSQL
  # from app.main.user.infrastructure.adapter.output.pg_user_storage import PgUserStorage
  # repository = PgUserStorage()
  
  # For In Memory (default in the base template)
  from app.main.user.infrastructure.adapter.output.in_memory_user_repository import InMemoryUserRepository
  repository = InMemoryUserRepository()

  Change the commented lines to choose the implementation.

• Configuration (PostgreSQL - Local /.env):

  DATABASE_USERNAME=postgres
  DATABASE_PASSWORD=postgres
  DATABASE_NAME=hexa
  DATABASE_HOST=localhost
  DATABASE_PORT="5432"
  SCHEMA_NAME=public

• Implementation:

  • PostgreSQL: app/main/user/infrastructure/adapter/output/pg_user_storage.py, app/main/user/application/config/pg_connection.py, app/main/user/application/entities/user_entity.py.
  • In Memory: app/main/user/infrastructure/adapter/output/in_memory_user_repository.py.

• Disable PostgreSQL: Comment out PgUserStorage initialization and ensure InMemoryUserRepository is used. Dependencies sqlalchemy and psycopg2-binary are not needed if unused.

Kafka

• Description: Integration with Apache Kafka for message production and consumption.

• Configuration (Local /.env):

  # Confluent Cloud example:
  BOOTSTRAP_SERVER=pkc-xxxx.xxxx.gcp.confluent.cloud:9092
  CLUSTER_API_KEY=TU_KAFKA_API_KEY
  CLUSTER_API_SECRET=TU_KAFKA_API_SECRET
  # Local example:
  # BOOTSTRAP_SERVER=localhost:9092
  TOPIC_RECIEVER=tu_topic_kafka_consumidor # For KafkaConsumer
  # For KafkaNotifier (hardcoded in kafka.py, adjust if necessary)
  TOPIC_PRODUCER=tmpl_arq-dev-topic-test_topic_producer
  RETRIES=10 # For producer
  CONSUMER_GROUP_ID=tu_grupo_consumidor_kafka # For KafkaConsumer

  Configurations are read in app/main/message/application/configurations.py.

• Implementation:

  • Producer: app/main/message/infrastructure/kafka.py (KafkaNotifier class).
  • Consumer: app/main/user/infrastructure/adapter/input/kafka_receiver.py (KafkaConsumer class).
  • The consumer is started in main.py during the application lifecycle. The notifier is injected into UserServiceUserImpl.

• Disable:

  1. Comment out or remove the following sections in main.py:

     # {{ KAFKA }}
     # from app.main.message.infrastructure.kafka import KafkaNotifier
     # from app.main.user.infrastructure.adapter.input.kafka_receiver import KafkaConsumer
     # {{! KAFKA }}
     
     # ...
     # {{ KAFKA }}
     # notifier = KafkaNotifier() # If Kafka is the only active notifier
     # {{! KAFKA }}
     
     # ...
     # {{ KAFKA }}
     # kafka_consumer = KafkaConsumer()
     # {{! KAFKA }}
     
     # In app_lifespan:
     # {{ KAFKA }}
     # await asyncio.create_task(KafkaConsumer.receive_user(user_service))
     # {{! KAFKA }}

  2. If KafkaNotifier is the notifier assigned to user_service, assign None or another notifier.

  3. Remove the confluent-kafka dependency from requirements.txt and Pipfile.

Google Cloud Pub/Sub

• Description: Integration with Google Cloud Pub/Sub for message production and consumption.

• Configuration (Local /.env):

  # Used by PubsubConsumer (configurations.py)
  PROJECT_ID=tu-gcp-project-id
  SUBSCRIPTION_ID=tu-pubsub-subscription-id-para-consumir
  
  # Used by PubsubNotifier (main.py)
  PUBSUB_PROJECT_ID=tu-gcp-project-id # Can be the same as PROJECT_ID
  PUBSUB_TOPIC_ID=tu-pubsub-topic-id-para-producir
  
  # GCP Authentication (locally)
  GOOGLE_APPLICATION_CREDENTIALS=/ruta/a/tus/credenciales-gcp.json

  Consumer configuration is read in app/main/message/application/configurations.py. The producer is configured directly in main.py. Note: PUBSUB_PROJECT_ID and PUBSUB_TOPIC_ID in main.py are set as "${{...}}". For local execution, they must be replaced with environ.get("PUBSUB_PROJECT_ID") or provide values directly in main.py if templating is not used.

• Implementation:
  • Producer: app/main/message/infrastructure/pubsub.py (PubsubNotifier class).
  • Consumer: app/main/user/infrastructure/adapter/input/pubsub_receiver.py (PubsubConsumer class).
  • The consumer is started in main.py during the application lifecycle. The notifier is injected into UserServiceUserImpl.

• Disable:

  1. Comment out or remove the following sections in main.py:

     # {{ PUBSUB }}
     # from app.main.message.infrastructure.pubsub import PubsubNotifier
     # from app.main.user.infrastructure.adapter.input.pubsub_receiver import PubsubConsumer
     # {{! PUBSUB }}
     
     # ...
     # {{ PUBSUB }}
     # notifier = PubsubNotifier(environ.get("PUBSUB_PROJECT_ID"), environ.get("PUBSUB_TOPIC_ID")) # If Pub/Sub is the only active notifier
     # {{! PUBSUB }}
     
     # ...
     # {{ PUBSUB }}
     # pubsub_consumer = PubsubConsumer()
     # {{! PUBSUB }}
     
     # In app_lifespan:
     # {{ PUBSUB }}
     # await asyncio.create_task(PubsubConsumer.receive_user(user_service))
     # {{! PUBSUB }}

  2. If PubsubNotifier is the notifier assigned to user_service, assign None or another notifier.

  3. Remove the google-cloud-pubsub dependency from requirements.txt and Pipfile.

Testing

The project uses pytest for unit tests and coverage to measure code coverage.

Run Unit Tests

From the project root, within the virtual environment (pipenv shell):

python -m pytest -v --junitxml=reports/result.xml

• -v: Verbose mode.
• --junitxml=reports/result.xml: Generates a JUnit XML report in the reports folder.

Validate Coverage

1. Run tests with coverage: coverage can be integrated with pytest or run separately. To run with pytest-cov (if installed):

   pipenv run pytest --cov=app --cov-report=xml --cov-report=html

   Or using coverage directly:

   pipenv run coverage run -m pytest

2. View coverage report in the console:

   pipenv run coverage report

3. Generate HTML coverage report:

   pipenv run coverage html

   This will create an htmlcov/ folder with the navigable report.

Test files are located in the app/tests/ directory.

Security Considerations

• Secret Management:
  • Locally: Use the .env file to store credentials and sensitive keys. This file is included in .gitignore and must NOT be committed to the repository.
  • Deployment (GKE/Cloud Run): Secrets (such as database passwords, Kafka/PubSub API keys, GOOGLE_CLIENT_ID) must be managed through the environment's secret mechanisms (e.g. Kubernetes Secrets, Google Secret Manager) and injected as environment variables into containers. See deploy/gke/deployment.yaml.erb for examples of how secrets are referenced.
• API Authentication:
  • Header uuid: Endpoints in the user module currently require a non-empty uuid header. Its purpose and validation beyond existence are not specified and should be defined according to security requirements.
  • Google OAuth2 Token: The file app/main/auth/infrastructure/auth.py contains logic to validate Google ID tokens. It requires the GOOGLE_CLIENT_ID environment variable. This validation can be added as a dependency (Depends(validate_token)) to FastAPI endpoints that need to be protected.

• Non-Versioned Files: Ensure the following files and directories that may contain sensitive or environment-specific information are not versioned:
  • htmlcov/
  • reports/
  • .env
  • Pipfile (if it contains sensitive information, although it usually does not)
  • Pipfile.lock (for the same reasons as Pipfile)
  • .coverage