Workflow Step: collab → design
User Input: Test run for microservice_scaffold
This document outlines the comprehensive design specifications for a scaffolded microservice, focusing on architecture, technology choices, project structure, developer experience (DX), and documentation. The goal is to generate a robust, maintainable, and easily extensible microservice that adheres to modern best practices.
The scaffolded microservice will be designed with a clear separation of concerns, promoting modularity and maintainability.
For this test run, we will demonstrate a popular and modern stack. The actual scaffolder will allow for configurable choices.
pytest-asyncio for async tests).The generated project will include the following core directories and files:
src/: Main application source code.tests/: Unit and integration tests.docs/: Additional documentation and OpenAPI specification.docker/: Dockerfile and related containerization assets.cicd/: CI/CD pipeline configurations.deploy/: Deployment scripts and configurations (e.g., Kubernetes manifests, Helm charts).config/: Environment-specific configurations.migrations/: Database migration scripts (e.g., Alembic).README.md: Comprehensive project overview and instructions.pyproject.toml: Poetry configuration for dependencies.Makefile: Common development and deployment commands.This section describes the "layout" of the generated microservice, both in terms of its file system structure and its exposed API.
. ├── README.md # Project overview, setup, usage instructions ├── Makefile # Common commands (install, run, test, lint, build, deploy) ├── pyproject.toml # Poetry project and dependency configuration ├── poetry.lock # Locked dependencies ├── docker-compose.yml # Local development environment (app, db, etc.) ├── .env.example # Example environment variables ├── .gitignore # Git ignore rules ├── src/ # Main application source code │ ├── main.py # FastAPI application entry point, app initialization │ ├── api/ # API routes/endpoints definitions │ │ ├── __init__.py │ │ ├── v1/ # Versioned API (e.g., /api/v1) │ │ │ ├── __init__.py │ │ │ ├── items.py # CRUD operations for 'items' resource │ │ │ └── health.py # Health check endpoint │ │ └── schemas/ # Pydantic models for request/response bodies │ │ ├── __init__.py │ │ └── item.py # ItemCreate, ItemUpdate, ItemResponse schemas │ ├── service/ # Business logic layer │ │ ├── __init__.py │ │ └── item_service.py # Business logic for item operations │ ├── repository/ # Data access layer │ │ ├── __init__.py │ │ ├── item_repository.py # Database interactions for items │ │ └── database.py # Database connection, session management │ ├── models/ # SQLAlchemy ORM models │ │ ├── __init__.py │ │ └── item_model.py # SQLAlchemy model for Item │ ├── config/ # Application configuration │ │ ├── __init__.py │ │ └── settings.py # Pydantic-based settings management │ └── core/ # Core utilities, exceptions, middleware │ ├── __init__.py │ ├── exceptions.py │ └── middleware.py ├── tests/ # Tests for the microservice │ ├── unit/ # Unit tests (e.g., service, repository logic) │ │ ├── __init__.py │ │ └── test_item_service.py │ ├── integration/ # Integration tests (e.g., API endpoints, DB interactions) │ │ ├── __init__.py │ │ └── test_item_api.py │ └── conftest.py # Pytest fixtures ├── docs/ # Additional documentation │ ├── openapi.yaml # Generated OpenAPI spec (or manually maintained) │ └── usage.md # Detailed usage guide ├── migrations/ # Database migrations (e.g., Alembic scripts) │ ├── env.py │ ├── script.py.mako │ └── versions/ # Migration files │ └── <timestamp>_initial.py ├── .github/ # GitHub Actions CI/CD workflows │ └── workflows/ │ ├── ci.yml # Lint, test, build Docker image │ └── cd.yml # Optional: Deploy to staging/production ├── deploy/ # Deployment configurations │ ├── kubernetes/ # Kubernetes manifests │ │ ├── deployment.yaml # K8s Deployment for the service │ │ ├── service.yaml # K8s Service │ │ └── ingress.yaml # K8s Ingress │ └── helm/ # Helm chart structure │ ├── Chart.yaml │ ├── values.yaml │ └── templates/ │ ├── _helpers.tpl │ ├── deployment.yaml │ └── service.yaml
The generated microservice will include a sample CRUD (Create, Read, Update, Delete) resource, for instance, Item, to demonstrate common API patterns.
Base URL: /api/v1
| HTTP Method | Endpoint | Description | Request Body (Example) | Response Body (Example) | Status |
| :---------- | :----------------- | :--------------------------------------------- | :--------------------------------- | :---------------------------------------------------- | :----- |
| GET | /items | Retrieve a list of all items. | N/A | [{"id": "uuid", "name": "Item A", "description": "Desc A"}] | 200 OK |
| GET | /items/{item_id} | Retrieve a specific item by ID. | N/A | {"id": "uuid", "name": "Item A", "description": "Desc A"} | 200 OK |
| POST | /items | Create a new item. | {"name": "New Item", "description": "New Description"} | {"id": "uuid", "name": "New Item", "description": "New Description"} | 201 Created |
| PUT | /items/{item_id} | Update an existing item by ID. | {"name": "Updated Item"} | {"id": "uuid", "name": "Updated Item", "description": "New Description"} | 200 OK |
| DELETE | /items/{item_id} | Delete an item by ID. | N/A | { "message": "Item deleted successfully" } | 200 OK |
| GET | /health | Health check endpoint for liveness/readiness. | N/A | {"status": "healthy"} | 200 OK |
Error Responses (Common):
400 Bad Request: Invalid input data.404 Not Found: Resource not found.422 Unprocessable Entity: Validation errors (FastAPI default).500 Internal Server Error: Server-side error.While microservices are primarily backend, the "design" extends to the readability and aesthetic of generated documentation and developer-facing tools.
* Primary Accent: #60A5FA (blue, for links, buttons).
* Text/Background: Default light theme (black text on white/light grey background).
* HTTP Methods: Standard color coding (GET-green, POST-blue, PUT-orange, DELETE-red).
Makefile commands, CI/CD logs, and local scripts. * Success: Green (e.g., [SUCCESS] Tests passed.)
* Failure/Error: Red (e.g., [ERROR] Build failed.)
* Warning: Yellow (e.g., [WARNING] Configuration missing.)
* Information/Highlight: Cyan or Blue (e.g., [INFO] Starting service...)
* Neutral/Default: White or Grey for general output.
collab → generate_code - Microservice ScaffoldingThis deliverable provides a comprehensive microservice scaffold based on your request for a "Test run for microservice_scaffold". We have generated a fully functional boilerplate for a Product Service, demonstrating best practices for a modern, scalable microservice architecture.
The chosen technology stack for this scaffold is:
This output includes the complete project structure, source code, Docker setup, testing suite, CI/CD configuration, and deployment scripts, along with a detailed README.md for quick setup and understanding.
The generated project adheres to a clean, modular structure designed for maintainability and scalability.
microservice-product-service/
├── .github/
│ └── workflows/
│ └── main.yml # GitHub Actions CI/CD pipeline
├── app/
│ ├── api/
│ │ └── v1/
│ │ └── endpoints/
│ │ └── products.py # FastAPI endpoints for Product resource
│ ├── crud/
│ │ └── products.py # CRUD operations for Product
│ ├── core/
│ │ ├── config.py # Application configuration settings
│ │ ├── database.py # Database session management
│ │ └── exceptions.py # Custom application exceptions
│ ├── models/
│ │ └── product.py # SQLAlchemy ORM model for Product
│ ├── schemas/
│ │ └── product.py # Pydantic schemas for request/response validation
│ └── main.py # Main FastAPI application entry point
├── kubernetes/
│ ├── deployment.yaml # Kubernetes Deployment manifest
│ └── service.yaml # Kubernetes Service manifest
├── scripts/
│ └── deploy_to_vm.sh # Example script for VM deployment
├── tests/
│ ├── api/
│ │ └── v1/
│ │ └── test_products.py # Unit/Integration tests for Product API
│ └── conftest.py # Pytest fixtures and configurations
├── .dockerignore # Files/directories to ignore in Docker build context
├── .env.example # Example environment variables
├── Dockerfile # Docker build instructions for the application
├── docker-compose.yml # Docker Compose for local development (app + db)
├── Makefile # Common development commands
├── pyproject.toml # Poetry configuration for dependencies (or requirements.txt)
├── README.md # Project documentation and setup guide
└── requirements.txt # Python dependencies
app/main.py - FastAPI Application Entry Point
# app/main.py
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.responses import RedirectResponse
from sqlalchemy.orm import Session
from app.api.v1.endpoints import products
from app.core.config import settings
from app.core.database import engine, Base, get_db
from app.core.exceptions import ItemNotFoundException
import logging
# Configure basic logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# Create all database tables (if they don't exist)
# In a production environment, you'd use Alembic for migrations.
Base.metadata.create_all(bind=engine)
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.API_V1_STR,
description="A microservice for managing product catalog.",
openapi_url=f"{settings.API_V1_STR}/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
)
# Include API routers
app.include_router(products.router, prefix=settings.API_V1_STR, tags=["products"])
@app.get("/", include_in_schema=False)
async def root():
"""
Redirects to the API documentation.
"""
return RedirectResponse(url="/docs")
@app.get("/health", response_model=dict, status_code=status.HTTP_200_OK)
async def health_check(db: Session = Depends(get_db)):
"""
Health check endpoint to verify service and database connectivity.
"""
try:
# Try to execute a simple query to check DB connection
db.execute("SELECT 1")
logger.info("Health check successful: Application and database are operational.")
return {"status": "ok", "message": "Product Service is healthy and connected to DB."}
except Exception as e:
logger.error(f"Health check failed: Database connection error - {e}")
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"Database connection failed: {e}"
)
# Global exception handler for custom exceptions
@app.exception_handler(ItemNotFoundException)
async def item_not_found_exception_handler(request, exc: ItemNotFoundException):
"""
Handles ItemNotFoundException globally, returning a 404 Not Found.
"""
logger.warning(f"Item not found: {exc.detail}")
return HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=exc.detail
)
logger.info(f"FastAPI application '{settings.PROJECT_NAME}' initialized.")
app/api/v1/endpoints/products.py - Product API Routes
# app/api/v1/endpoints/products.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.schemas import product as product_schemas
from app.crud import products as product_crud
from app.core.database import get_db
from app.core.exceptions import ItemNotFoundException
import logging
router = APIRouter()
logger = logging.getLogger(__name__)
@router.post("/", response_model=product_schemas.Product, status_code=status.HTTP_201_CREATED)
def create_product(product: product_schemas.ProductCreate, db: Session = Depends(get_db)):
"""
Create a new product.
"""
logger.info(f"Attempting to create product with name: {product.name}")
db_product = product_crud.get_product_by_name(db, name=product.name)
if db_product:
logger.warning(f"Product with name '{product.name}' already exists.")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Product with this name already exists"
)
new_product = product_crud.create_product(db=db, product=product)
logger.info(f"Product created successfully with ID: {new_product.id}")
return new_product
@router.get("/", response_model=List[product_schemas.Product])
def read_products(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
"""
Retrieve a list of products.
"""
logger.debug(f"Fetching products with skip={skip}, limit={limit}")
products = product_crud.get_products(db, skip=skip, limit=limit)
return products
@router.get("/{product_id}", response_model=product_schemas.Product)
def read_product(product_id: int, db: Session = Depends(get_db)):
"""
Retrieve a single product by its ID.
"""
logger.debug(f"Fetching product with ID: {product_id}")
try:
product = product_crud.get_product(db, product_id=product_id)
except ItemNotFoundException as e:
logger.warning(f"Product with ID {product_id} not found.")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=e.detail)
return product
@router.put("/{product_id}", response_model=product_schemas.Product)
def update_product(product_id: int, product: product_schemas.ProductUpdate, db: Session = Depends(get_db)):
"""
Update an existing product.
"""
logger.info(f"Attempting to update product with ID: {product_id}")
try:
updated_product = product_crud.update_product(db, product_id=product_id, product_in=product)
logger.info(f"Product with ID {product_id} updated successfully.")
return updated_product
except ItemNotFoundException as e:
logger.warning(f"Product with ID {product_id} not found for update.")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=e.detail)
@router.delete("/{product_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_product(product_id: int, db: Session = Depends(get_db)):
"""
Delete a product by its ID.
"""
logger.info(f"Attempting to delete product with ID: {product_id}")
try:
product_crud.delete_product(db, product_id=product_id)
logger.info(f"Product with ID {product_id} deleted successfully.")
return {"message": "Product deleted successfully"}
except ItemNotFoundException as e:
logger.warning(f"Product with ID {product_id} not found for deletion.")
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=e.detail)
app/crud/products.py - CRUD Operations
# app/crud/products.py
from typing import List, Optional
from sqlalchemy.orm import Session
from app.models import product as product_model
from app.schemas import product as product_schemas
from app.core.exceptions import ItemNotFoundException
import logging
logger = logging.getLogger(__name__)
def get_product(db: Session, product_id: int) -> Optional[product_model.Product]:
"""
Retrieve a product by its ID.
Raises ItemNotFoundException if product does not exist.
"""
product = db.query(product_model.Product).filter(product_model.Product.id == product_id).first()
if not product:
logger.debug(f"Product with ID {product_id} not found in DB.")
raise ItemNotFoundException(detail=f"Product with ID {product_id} not found")
return product
def get_product_by_name(db: Session, name: str) -> Optional[product_model.Product]:
"""
Retrieve a product by its name.
"""
return db.query(product_model.Product).filter(product_model.Product.name == name).first()
def get_products(db: Session, skip: int = 0, limit: int = 100) -> List[product_model.Product]:
"""
Retrieve a list of products with pagination.
"""
return db.query(product_model.Product).offset(skip).limit(limit).all()
def create_product(db: Session, product: product_schemas.ProductCreate) -> product_model.Product:
"""
Create a new product in the database.
"""
db_product = product_model.Product(
name=product.name,
description=product.description,
price=product.price,
stock=product.stock
)
db.add(db_product)
db.commit()
db.refresh(db_product)
logger.info(f"Product '{db_product.name}' created in DB.")
return db_product
def update_product(db: Session, product_id: int, product_in: product_schemas.ProductUpdate) -> product_model.Product:
"""
Update an existing product by its ID.
Raises ItemNotFoundException if product does not exist.
"""
db_product = get_product(db, product_id) # Reuses get_product for existence check
update_data = product_in.dict(exclude_unset=True) # Only update fields that are provided
for field, value in update_data.items():
setattr(db_product, field, value)
db.add(db_product)
db.commit()
db.refresh(db_product)
logger.info(f"Product with ID {product_id} updated in DB.")
This is a comprehensive deliverable for the "Microservice Scaffolder" workflow, specifically for the generate_code step based on your "Test run for microservice_scaffold" request.
We have generated a complete, production-ready microservice scaffolding for a Product Service using Python (Flask), PostgreSQL, Docker, Pytest, and GitHub Actions for CI/CD. This output includes all requested components: API routes, database models, tests, Docker setup, CI/CD pipeline configuration, and a basic deployment script.
This deliverable provides the complete code and configuration for a sample Product Service microservice. This service manages product information (ID, name, description, price, stock) and demonstrates a typical microservice architecture.
Key Technologies Used:
The generated microservice adheres to a standard, maintainable project structure:
microservice_scaffold_test/
├── app/
│ ├── __init__.py # Initializes Flask app, DB, and registers blueprints
│ ├── app.py # Main application entry point (WSGI server)
│ ├── config.py # Application configuration settings
│ ├── models.py # SQLAlchemy database models
│ ├── routes.py # API route definitions (Blueprint)
│ └── services.py # Business logic for product operations
├── tests/
│ ├── conftest.py # Pytest fixtures for testing setup
│ ├── test_models.py # Unit tests for database models
│ └── test_routes.py # Integration tests for API endpoints
├── .github/
│ └── workflows/
│ └── ci-cd.yml # GitHub Actions CI/CD pipeline configuration
├── Dockerfile # Docker build instructions for the application
├── docker-compose.yml # Docker Compose for local development (app + db)
├── requirements.txt # Python dependencies
├── deploy.sh # Basic deployment script
└── README.md # Project README and setup instructions
Below are the contents of each generated file, along with explanations.
README.md
# Product Service Microservice (Test Run)
This repository contains a scaffolded Python Flask microservice for managing products. It includes API endpoints for CRUD operations, a PostgreSQL database model, Docker setup for local development, unit/integration tests, and a CI/CD pipeline configuration using GitHub Actions.
## Features
* **RESTful API**: Endpoints for creating, reading, updating, and deleting products.
* **Database**: PostgreSQL with SQLAlchemy ORM.
* **Containerization**: Docker and Docker Compose for easy setup and deployment.
* **Testing**: Pytest for unit and integration tests.
* **CI/CD**: GitHub Actions workflow for automated build, test, and (placeholder) deployment.
## Local Development Setup
### Prerequisites
* Docker and Docker Compose
* Python 3.9+ (optional, for running locally without Docker)
### 1. Clone the repository
git clone https://github.com/your-org/microservice_scaffold_test.git
cd microservice_scaffold_test
### 2. Build and Run with Docker Compose
This will set up the PostgreSQL database and run the Flask application.
docker-compose up --build
The application will be accessible at `http://localhost:5000`.
### 3. Initialize Database (first run)
While the services are running, execute the following to create the database tables:
docker-compose exec app python -c "from app import create_app, db; app = create_app(); app.app_context().push(); db.create_all()"
You should see output indicating the Flask application is running.
### 4. Run Tests
docker-compose exec app pytest
### 5. API Endpoints (Example with `curl`)
* **Create Product**:
curl -X POST -H "Content-Type: application/json" -d '{"name": "Laptop", "description": "Powerful gaming laptop", "price": 1200.00, "stock": 50}' http://localhost:5000/products
* **Get All Products**:
curl http://localhost:5000/products
* **Get Product by ID (e.g., ID=1)**:
curl http://localhost:5000/products/1
* **Update Product (e.g., ID=1)**:
curl -X PUT -H "Content-Type: application/json" -d '{"name": "Gaming Laptop", "description": "High-performance gaming laptop", "price": 1250.00, "stock": 45}' http://localhost:5000/products/1
* **Delete Product (e.g., ID=1)**:
curl -X DELETE http://localhost:5000/products/1
## CI/CD Pipeline
The `.github/workflows/ci-cd.yml` file defines a GitHub Actions pipeline that:
1. Lints the code.
2. Runs all tests.
3. (Placeholder) Builds a Docker image and pushes it to a container registry.
4. (Placeholder) Deploys the service.
## Deployment
Refer to `deploy.sh` for a basic example of building and pushing a Docker image. Actual deployment strategies will vary based on your cloud provider (AWS ECS/EKS, Kubernetes, etc.).
requirements.txt
Flask==2.3.2
Flask-SQLAlchemy==3.0.3
psycopg2-binary==2.9.6
python-dotenv==1.0.0
pytest==7.4.0
pytest-flask==1.3.0
Explanation: Lists all Python dependencies required for the application and testing.
app/config.py
import os
from dotenv import load_dotenv
load_dotenv() # Load environment variables from .env file
class Config:
"""Base configuration for the application."""
DEBUG = False
TESTING = False
SECRET_KEY = os.environ.get('SECRET_KEY', 'a_secret_key_for_dev')
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'postgresql://user:password@db:5432/products_db')
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL', 'postgresql://user:password@db:5432/products_db')
class TestingConfig(Config):
"""Testing configuration."""
TESTING = True
# Use an in-memory SQLite database for faster, isolated tests
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
PRESERVE_CONTEXT_ON_EXCEPTION = False # Important for testing
class ProductionConfig(Config):
"""Production configuration."""
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') # Must be set in production
if SQLALCHEMY_DATABASE_URI is None:
raise ValueError("DATABASE_URL must be set for ProductionConfig")
Explanation: Defines different configuration classes for development, testing, and production environments. It loads environment variables from a .env file and sets the database URI, secret key, and debug mode.
app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from app.config import DevelopmentConfig, TestingConfig, ProductionConfig
import os
db = SQLAlchemy()
def create_app(config_class=DevelopmentConfig):
"""
Factory function to create and configure the Flask application.
"""
app = Flask(__name__)
app.config.from_object(config_class)
db.init_app(app)
# Register blueprints
from app.routes import products_bp
app.register_blueprint(products_bp, url_prefix='/')
# Optional: A simple health check route
@app.route('/health', methods=['GET'])
def health_check():
return {"status": "healthy"}, 200
return app
Explanation: This is the application factory. It initializes the Flask app, configures it based on the provided config_class, initializes SQLAlchemy, and registers the products_bp blueprint for API routes. It also includes a basic health check endpoint.
app/models.py
from app import db
class Product(db.Model):
"""
Represents a product in the database.
"""
__tablename__ = 'products' # Explicitly define table name
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(100), nullable=False)
description = db.Column(db.Text, nullable=True)
price = db.Column(db.Float, nullable=False)
stock = db.Column(db.Integer, nullable=False, default=0)
created_at = db.Column(db.DateTime, default=db.func.current_timestamp())
updated_at = db.Column(db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp())
def __repr__(self):
return f'<Product {self.name}>'
def to_dict(self):
"""Converts the Product object to a dictionary."""
return {
'id': self.id,
'name': self.name,
'description': self.description,
'price': self.price,
'stock': self.stock,
'created_at': self.created_at.isoformat() if self.created_at else None,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
@staticmethod
def from_dict(data):
"""Creates a Product object from a dictionary."""
product = Product(
name=data.get('name'),
description=data.get('description'),
price=data.get('price'),
stock=data.get('stock', 0)
)
return product
Explanation: Defines the Product SQLAlchemy model, mapping it to a products table in the database. It includes fields for ID, name, description, price, stock, and timestamps. to_dict and from_dict methods are provided for easy serialization/deserialization.
app/services.py
from app import db
from app.models import Product
class ProductService:
"""
Encapsulates business logic for Product operations.
"""
@staticmethod
def get_all_products():
"""Retrieves all products from the database."""
return Product.query.all()
@staticmethod
def get_product_by_id(product_id):
"""Retrieves a single product by its ID."""
return Product.query.get(product_id)
@staticmethod
def create_product(data):
"""Creates a new product."""
if not all(k in data for k in ['name', 'price']):
raise ValueError("Name and price are required to create a product.")
product = Product.from_dict(data)
db.session.add(product)
db.session.commit()
return product
@staticmethod
def update_product(product_id, data):
"""Updates an existing product."""
product = Product.query.get(product_id)
if not product:
return None
# Update fields if present in data
product.name = data.get('name', product.name)
product.description = data.get('description', product.description)
product.price = data.get('price', product.price)
product.stock = data.get('stock', product.stock)
db.session.commit()
return product
@staticmethod
def delete_product(product_id
Congratulations! Your request for a microservice scaffold has been successfully processed. We've generated a comprehensive, production-ready foundation for your new microservice, "microservice_scaffold", complete with essential components for development, testing, deployment, and operation. This deliverable provides a robust starting point, allowing your team to focus on business logic rather than boilerplate.
This scaffold provides a Python-based microservice using FastAPI for its high performance and developer-friendliness, SQLAlchemy for ORM with a PostgreSQL database, and full Docker integration. It includes a structured project layout, API routes, database models, testing setup, and ready-to-use CI/CD configurations.
Below is the directory structure for your new microservice. Each component is designed for clarity, scalability, and maintainability.
microservice_scaffold/
├── .github/
│ └── workflows/
│ └── main.yml # GitHub Actions CI/CD pipeline
├── app/
│ ├── api/
│ │ └── v1/
│ │ └── endpoints/
│ │ └── items.py # Example API endpoints for 'items'
│ ├── core/
│ │ ├── config.py # Application settings and environment variables
│ │ └── database.py # Database connection and session management
│ ├── crud/
│ │ └── crud_item.py # CRUD operations for 'Item' model
│ ├── models/
│ │ └── item.py # SQLAlchemy database model for 'Item'
│ ├── schemas/
│ │ └── item.py # Pydantic schemas for request/response validation
│ └── main.py # FastAPI application entry point
├── tests/
│ ├── conftest.py # Pytest fixtures for testing
│ └── test_items.py # Example unit/integration tests for 'items' API
├── Dockerfile # Defines the Docker image for the microservice
├── docker-compose.yml # Local development setup with PostgreSQL
├── .dockerignore # Files to ignore when building Docker image
├── requirements.txt # Production Python dependencies
├── requirements-dev.txt # Development Python dependencies (including testing tools)
├── .env.example # Example environment variables
├── README.md # Project documentation and setup instructions
└── deploy.sh # Example deployment script
app/)app/main.py: The heart of your FastAPI application. It initializes the app, includes routers, and sets up event handlers (like database connection on startup/shutdown).
# app/main.py
from fastapi import FastAPI
from app.core.config import settings
from app.core.database import engine, Base
from app.api.v1.endpoints import items
import uvicorn
# Create database tables
Base.metadata.create_all(bind=engine)
app = FastAPI(
title=settings.PROJECT_NAME,
version=settings.API_VERSION,
openapi_url=f"{settings.API_V1_STR}/openapi.json"
)
app.include_router(items.router, prefix=settings.API_V1_STR, tags=["items"])
@app.get("/")
async def root():
return {"message": "Welcome to microservice_scaffold!"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
app/api/v1/endpoints/items.py: Demonstrates basic RESTful API endpoints for managing "items".
# app/api/v1/endpoints/items.py
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.schemas.item import ItemCreate, ItemResponse
from app.crud.crud_item import crud_item
from app.core.database import get_db
router = APIRouter()
@router.post("/items/", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
db_item = crud_item.create(db, obj_in=item)
return db_item
@router.get("/items/", response_model=List[ItemResponse])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud_item.get_multi(db, skip=skip, limit=limit)
return items
@router.get("/items/{item_id}", response_model=ItemResponse)
def read_item(item_id: int, db: Session = Depends(get_db)):
item = crud_item.get(db, id=item_id)
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return item
app/models/item.py: Defines the SQLAlchemy ORM model for an Item.
# app/models/item.py
from sqlalchemy import Column, Integer, String, Boolean
from app.core.database import Base
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
description = Column(String, nullable=True)
is_active = Column(Boolean, default=True)
app/schemas/item.py: Pydantic models for request body validation and response serialization.
# app/schemas/item.py
from pydantic import BaseModel, Field
class ItemBase(BaseModel):
name: str = Field(..., example="My Awesome Item")
description: str | None = Field(None, example="A detailed description of the item.")
is_active: bool = True
class ItemCreate(ItemBase):
pass # Can add specific fields for creation if needed
class ItemResponse(ItemBase):
id: int = Field(..., example=1)
class Config:
from_attributes = True # for SQLAlchemy integration
app/crud/crud_item.py: Generic CRUD operations for the Item model, promoting code reusability.
# app/crud/crud_item.py
from sqlalchemy.orm import Session
from app.models.item import Item
from app.schemas.item import ItemCreate, ItemResponse
from app.crud.base import CRUDBase
class CRUDItem(CRUDBase[Item, ItemCreate, ItemResponse]):
pass
crud_item = CRUDItem(Item)
Dockerfile: Defines how to build your microservice's Docker image.
# Dockerfile
FROM python:3.11-slim-buster
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
docker-compose.yml: For local development, setting up the service and a PostgreSQL database.
# docker-compose.yml
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
env_file:
- .env
depends_on:
- db
command: uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
volumes:
- ./app:/app/app # Mount app directory for live reloading
db:
image: postgres:15-alpine
env_file:
- .env
volumes:
- postgres_data:/var/lib/postgresql/data/
volumes:
postgres_data:
tests/)tests/test_items.py: Example tests for your API endpoints using pytest and httpx.
# tests/test_items.py
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.main import app
from app.core.database import Base, get_db
from app.models.item import Item
import pytest
# Use a separate test database
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
@pytest.fixture(name="db_session")
def db_session_fixture():
Base.metadata.create_all(bind=engine) # Create tables
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
Base.metadata.drop_all(bind=engine) # Drop tables after tests
@pytest.fixture(name="client")
def client_fixture(db_session):
def override_get_db():
yield db_session
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as c:
yield c
app.dependency_overrides.clear()
def test_create_item(client):
response = client.post(
"/api/v1/items/",
json={"name": "Test Item", "description": "This is a test item"}
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "Test Item"
assert "id" in data
def test_read_items(client):
# First, create an item
client.post("/api/v1/items/", json={"name": "Item 1"})
client.post("/api/v1/items/", json={"name": "Item 2"})
response = client.get("/api/v1/items/")
assert response.status_code == 200
data = response.json()
assert len(data) == 2
assert data[0]["name"] == "Item 1"
.github/workflows/main.yml: A GitHub Actions workflow for building, testing, and deploying your microservice.
# .github/workflows/main.yml
name: CI/CD Pipeline
on:
push:
branches:
- main
pull_request:
branches:
- main
jobs:
build-and-test:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt -r requirements-dev.txt
- name: Run tests
run: |
pytest tests/
- name: Build Docker image
run: docker build -t microservice_scaffold:${{ github.sha }} .
- name: Log in to Docker Hub (or other registry)
uses: docker/login-action@v2
with:
username: ${{ secrets.DOCKER_USERNAME }}
password: ${{ secrets.DOCKER_PASSWORD }}
- name: Push Docker image
run: |
docker tag microservice_scaffold:${{ github.sha }} ${{ secrets.DOCKER_USERNAME }}/microservice_scaffold:${{ github.sha }}
docker push ${{ secrets.DOCKER_USERNAME }}/microservice_scaffold:${{ github.sha }}
deploy:
needs: build-and-test
runs-on: ubuntu-latest
environment: production
steps:
- name: Checkout code
uses: actions/checkout@v3
- name: Deploy to Server (e.g., via SSH or Kubernetes)
env:
SSH_PRIVATE_KEY: ${{ secrets.SSH_PRIVATE_KEY }}
SERVER_USER: ${{ secrets.SERVER_USER }}
SERVER_HOST: ${{ secrets.SERVER_HOST }}
DOCKER_IMAGE: ${{ secrets.DOCKER_USERNAME }}/microservice_scaffold:${{ github.sha }}
run: |
echo "$SSH_PRIVATE_KEY" > deploy_key
chmod 600 deploy_key
ssh -i deploy_key -o StrictHostKeyChecking=no $SERVER_USER@$SERVER_HOST '
docker pull $DOCKER_IMAGE && \
docker stop microservice_scaffold_app || true && \
docker rm microservice_scaffold_app || true && \
docker run -d --name microservice_scaffold_app -p 8000:8000 $DOCKER_IMAGE
'
deploy.sh: A simple example script for deploying to a remote server (can be adapted for Kubernetes, AWS ECS, etc.).
#!/bin/bash
# deploy.sh - Example deployment script
# --- Configuration ---
REMOTE_USER="your-ssh-user"
REMOTE_HOST="your-server-ip-or-hostname"
DOCKER_IMAGE_NAME="your-dockerhub-username/microservice_scaffold"
DOCKER_IMAGE_TAG="latest" # Or use a specific commit SHA/version
# --- Deployment Steps ---
echo "--- Building Docker image locally ---"
docker build -t ${DOCKER_IMAGE_NAME}:${DOCKER_IMAGE_TAG} .
echo "--- Pushing Docker image to registry ---"
docker push ${DOCKER_IMAGE_NAME}:${DOCKER_IMAGE_TAG}
echo "--- Deploying to remote host: ${REMOTE_HOST} ---"
ssh ${REMOTE_USER}@${REMOTE_HOST} << EOF
echo "Pull
\n