Building Microservices with Flask: Architecture and Best Practices

Flask’s simplicity makes it an excellent choice for microservices architecture. Its lightweight nature allows you to build focused, single-responsibility services that can scale independently.

Microservice Design Principles

Single Responsibility

Each service should have one clear business purpose:

# User Service
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://localhost/users'
db = SQLAlchemy(app)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), unique=True, nullable=False)
    name = db.Column(db.String(80), nullable=False)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    user = User(email=data['email'], name=data['name'])
    db.session.add(user)
    db.session.commit()
    return jsonify({'id': user.id, 'email': user.email})

Service Communication Patterns

HTTP REST Communication

import requests
from flask import current_app

class UserServiceClient:
    def __init__(self, base_url):
        self.base_url = base_url
    
    def get_user(self, user_id):
        response = requests.get(f"{self.base_url}/users/{user_id}")
        if response.status_code == 200:
            return response.json()
        return None
    
    def create_user(self, user_data):
        response = requests.post(f"{self.base_url}/users", json=user_data)
        return response.json()

Event-Driven Communication with Redis

import redis
import json
from flask import current_app

class EventPublisher:
    def __init__(self, redis_client):
        self.redis = redis_client
    
    def publish_event(self, event_type, data):
        event = {
            'type': event_type,
            'data': data,
            'timestamp': datetime.utcnow().isoformat()
        }
        self.redis.publish('events', json.dumps(event))

# Usage in your service
@app.route('/users', methods=['POST'])
def create_user():
    user = create_user_logic(request.get_json())
    
    # Publish event for other services
    event_publisher.publish_event('user.created', {
        'user_id': user.id,
        'email': user.email
    })
    
    return jsonify(user.to_dict())

Configuration Management

import os
from dataclasses import dataclass

@dataclass
class Config:
    DATABASE_URL: str = os.getenv('DATABASE_URL')
    REDIS_URL: str = os.getenv('REDIS_URL')
    SERVICE_NAME: str = os.getenv('SERVICE_NAME', 'user-service')
    SERVICE_PORT: int = int(os.getenv('SERVICE_PORT', 5000))
    
    # Service discovery
    CONSUL_HOST: str = os.getenv('CONSUL_HOST', 'localhost')
    CONSUL_PORT: int = int(os.getenv('CONSUL_PORT', 8500))

config = Config()

Health Checks and Monitoring

@app.route('/health')
def health_check():
    try:
        # Check database connection
        db.session.execute('SELECT 1')
        
        # Check external dependencies
        redis_client.ping()
        
        return jsonify({
            'status': 'healthy',
            'service': config.SERVICE_NAME,
            'timestamp': datetime.utcnow().isoformat()
        })
    except Exception as e:
        return jsonify({
            'status': 'unhealthy',
            'error': str(e)
        }), 503

Service Discovery with Consul

import consul

class ServiceRegistry:
    def __init__(self, consul_host, consul_port):
        self.consul = consul.Consul(host=consul_host, port=consul_port)
    
    def register_service(self, name, port, health_check_url):
        self.consul.agent.service.register(
            name=name,
            service_id=f"{name}-{port}",
            port=port,
            check=consul.Check.http(health_check_url, interval="10s")
        )
    
    def discover_service(self, service_name):
        services = self.consul.health.service(service_name, passing=True)[1]
        if services:
            service = services[0]['Service']
            return f"http://{service['Address']}:{service['Port']}"
        return None

Docker Deployment

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 5000
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "app:app"]

Docker Compose for Development

version: '3.8'
services:
  user-service:
    build: ./user-service
    ports:
      - "5001:5000"
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/users
      - REDIS_URL=redis://redis:6379
  
  order-service:
    build: ./order-service
    ports:
      - "5002:5000"
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/orders
      - USER_SERVICE_URL=http://user-service:5000
  
  db:
    image: postgres:13
    environment:
      - POSTGRES_PASSWORD=password
  
  redis:
    image: redis:alpine

Flask microservices offer the perfect balance of simplicity and power. By following these patterns, you can build scalable, maintainable systems that grow with your business needs.