AppGain Servers - Microservice Development Blueprint¶

Table of Contents¶

Project Architecture Overview
Core Framework & Dependencies
Microservice Structure
API Design Patterns
Data Models & Database
Error Handling & Validation
Logging & Monitoring
Authentication & Security
Testing & Quality Assurance
Deployment & Configuration
Code Style & Standards
Example Microservice Template

Project Architecture Overview¶

Architecture Pattern¶

Type: Monolithic Flask application with modular microservice architecture
Pattern: Blueprint-based service separation
Database: MongoDB with MongoFrames ORM
Caching: Redis
Monitoring: Prometheus + Sentry
Deployment: Docker + Vercel integration

Core Principles¶

Separation of Concerns: Each microservice handles specific business domain
Consistent API Patterns: Standardized HTTP methods and response formats
Database Abstraction: MongoFrames for MongoDB operations
Error Handling: Centralized error management with custom error classes
Logging: Structured logging with Loki integration

Core Framework & Dependencies¶

Primary Framework¶

Flask==0.11.1  # Core web framework
MongoFrames==1.3.6  # MongoDB ORM
pymongo==4.6.1  # MongoDB driver

Key Dependencies¶

# Data Processing
arrow==0.8.0  # Date/time handling
simplejson==3.8.2  # JSON serialization
jsonschema==2.5.1  # JSON validation

# Monitoring & Logging
prometheus-flask-exporter==0.23.0  # Metrics
sentry-sdk==1.38.0  # Error tracking
python-logging-loki==0.3.1  # Log aggregation

# External Services
stripe==1.53.0  # Payment processing
boto3==1.33.13  # AWS services
redis==3.5.3  # Caching

# Development Tools
black==23.3.0  # Code formatting
pytest==7.4.4  # Testing

Microservice Structure¶

Directory Organization¶

admin_server/
├── __init__.py                 # Version and path configuration
├── app.py                      # Main Flask application
├── api/                        # Core API infrastructure
│   ├── blueprint.py            # Custom AppBlueprint class
│   ├── errors.py               # Error handling classes
│   └── helpers.py              # Common utility functions
├── utils/                       # Shared utilities
│   ├── logging.py              # Logging configuration
│   └── validation.py           # Data validation helpers
├── {service_name}/             # Microservice directory
│   ├── __init__.py             # Service initialization
│   ├── {service_name}.py       # Data models
│   ├── {service_name}_api.py   # API endpoints
│   └── {service_name}_thread.py # Background tasks (if needed)
└── templates/                   # HTML templates (if needed)

Naming Conventions¶

Service Directory: app_{service} or {service}_api
API File: {service_name}_api.py
Model File: {service_name}.py
Blueprint Variable: {service_name}_api or {service_name}_bp

API Design Patterns¶

Blueprint Registration¶

# Standard pattern
{service_name}_api = AppBlueprint("{service_name}", __name__)
api = {service_name}_api

# With URL prefix (if needed)
{service_name}_bp = AppBlueprint("{service_name}", __name__, url_prefix="/api/{service}")

# Registration in main app.py
from .{service_name}.{service_name}_api import {service_name}_api
app.register_blueprint({service_name}_api)

Route Definition Pattern¶

@api.{HTTP_METHOD}("/{endpoint}")
@api.accepts(["application/json"])
@api.validate_json(Model._json_schema)  # Optional
def {function_name}(parameters):
    """
    Docstring with clear description

    :param parameter: Description
    :return: Return description
    :rtype: Return type
    """
    try:
        # Implementation logic
        return make_response_json(
            payload_obj=result_data,
            status_code=200
        )
    except Exception as e:
        # Error handling
        raise e

HTTP Method Decorators¶

@api.GET("/endpoint")           # GET requests
@api.POST("/endpoint")          # POST requests
@api.PUT("/endpoint")           # PUT requests
@api.DELETE("/endpoint")        # DELETE requests

Route Parameter Format¶

# ✅ CORRECT - Use simple parameter names
@api.GET("/items/<item_id>")
@api.PUT("/users/<user_id>/status")

# ❌ INCORRECT - Don't use type annotations
@api.GET("/items/<string:item_id>")
@api.PUT("/users/<int:user_id>/status")

Data Models & Database¶

MongoFrames Model Structure¶

from mongoframes import Frame, Q
from bson import ObjectId
from ..utils.logging import get_logger

logger = get_logger(__name__)

class ServiceModel(Frame):
    """
    Service model description
    """

    _collection = "collection_name"

    _fields = {
        "field1",
        "field2",
        "field3",
        # ... other fields
    }

    _json_schema = {
        "title": "Model:Create",
        "type": "object",
        "required": ["required_field1", "required_field2"],
        "additionalProperties": False,
        "properties": {
            "field1": {
                "type": "string",
                "pattern": "^[\\w ]+$",
                "minLength": 5,
                "maxLength": 40,
            },
            "field2": {
                "type": "string",
                "enum": ["value1", "value2", "value3"]
            }
        }
    }

    def for_json(self):
        """Convert model to JSON serializable format"""
        return {
            "id": str(self._id),
            "field1": self.field1,
            "field2": self.field2,
            # ... other fields
        }

Database Operations Pattern¶

# Query operations
items = Model.many(Q.field == value)
item = Model.one(Q._id == ObjectId(id))
count = Model.count(Q.status == "active")

# Create operations
new_item = Model(data_dict)
result = new_item.insert()

# Update operations
item.field = new_value
item.update()

# Delete operations
item.delete()

# Complex queries
query = And(Q.status == "active", Q.type == "premium")
results = Model.many(query, skip=offset, limit=limit)

Error Handling & Validation¶

Custom Error Classes¶

from ..api.errors import APIError

# Standard error responses
raise APIError.BadRequestError("Invalid input data")
raise APIError.NotFoundError("Resource not found")
raise APIError.ForbiddenError("Access denied")
raise APIError.ServerError("Internal server error")
raise APIError.ConflictError("Resource conflict")

Error Response Format¶

# Standard error response structure
{
    "status": "failed",
    "message": "Error description",
    "error": "Detailed error information",
    "status_code": 400
}

JSON Validation¶

@api.validate_json(Model._json_schema)
def create_item():
    # Function will automatically validate JSON against schema
    pass

Request Validation Pattern¶

def create_item():
    try:
        data = request.get_json()
        if not data:
            raise APIError.BadRequestError("No data provided")

        # Validate required fields
        required_fields = ["name", "email"]
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            raise APIError.BadRequestError(f"Missing fields: {missing_fields}")

        # Process data
        item = Model(data)
        result = item.insert()

        return make_response_json(
            payload_obj={"message": "Item created", "id": str(result._id)},
            status_code=201
        )
    except Exception as e:
        raise e

Logging & Monitoring¶

Logger Setup¶

from ..utils.logging import get_logger

logger = get_logger(__name__)

# Usage patterns
logger.info("Operation completed successfully")
logger.warning("Potential issue detected")
logger.error("Error occurred: %s", str(error))
logger.debug("Debug information: %s", debug_data)

Structured Logging¶

# Log with context
logger.info("Processing user %s for suit %s", user_id, suit_id)

# Log with structured data
logger.info("API request processed", extra={
    "user_id": user_id,
    "suit_id": suit_id,
    "operation": "create",
    "duration_ms": duration
})

Monitoring Integration¶

# Prometheus metrics (automatic with prometheus-flask-exporter)
# Sentry error tracking (automatic with sentry-sdk)
# Loki log aggregation (automatic with python-logging-loki)

Authentication & Security¶

Authentication Decorator¶

from ..users.auth import Auth

@api.POST("/protected-endpoint")
@Auth.auth  # Requires authentication
def protected_function():
    # Function only accessible to authenticated users
    pass

User Context Access¶

from flask import g

def authenticated_function():
    # Access authenticated user information
    user_id = g.user_id
    user_role = g.user_role

    # Check permissions
    if user_role != "admin":
        raise APIError.ForbiddenError("Admin access required")

Testing & Quality Assurance¶

Testing Framework¶

# Use pytest for testing
# Test files should be in tests/ directory
# Follow naming convention: test_{service_name}.py

Code Quality Tools¶

# Black formatter for consistent code style
black {file_path}

# Pytest for testing
pytest tests/

# Coverage reporting
pytest --cov=admin_server tests/

Deployment & Configuration¶

Environment Configuration¶

# Configuration is loaded from environment variable
app.config.from_envvar("ADMIN_SERVER_CONFIG")

# Required environment variables
ADMIN_SERVER_CONFIG=config_file_path
MONGODB_URI=mongodb_connection_string
SECRET_KEY=application_secret_key
REDIS_HOST=redis_host
SENTRY_DSN=sentry_dsn

Docker Integration¶

# Dockerfile should include:
# - Python 3.5+ compatibility
# - All requirements.txt dependencies
# - Proper working directory setup
# - Health check endpoints

Code Style & Standards¶

Python Version Compatibility¶

Target: Python 3.5+ (based on shebang in app.py)
Encoding: Always use # -*- coding: utf-8 -*-
String Formatting: Use .format() method (not f-strings for Python 3.5 compatibility)

Import Organization¶

# Standard library imports
import os
import sys
from datetime import datetime

# Third-party imports
from flask import Flask, request
from mongoframes import Frame, Q

# Local imports
from ..utils.logging import get_logger
from ..api.errors import APIError

Documentation Standards¶

def function_name(param1, param2):
    """
    Clear function description

    :param param1: Parameter description
    :type param1: str
    :param param2: Parameter description
    :type param2: int
    :return: Return description
    :rtype: dict
    :raises: APIError.BadRequestError
    """
    pass

Example Microservice Template¶

Complete Microservice Structure¶

1. Service Directory Structure¶

app_example/
├── __init__.py
├── example.py              # Data model
├── example_api.py          # API endpoints
└── example_thread.py       # Background tasks (optional)

2. Data Model (example.py)¶

# -*- coding: utf-8 -*-
import arrow
from mongoframes import Frame, Q
from bson import ObjectId
from ..utils.logging import get_logger

logger = get_logger(__name__)

class Example(Frame):
    """
    Example service model
    """

    _collection = "examples"

    _fields = {
        "name",
        "description",
        "status",
        "created_at",
        "updated_at",
        "user_id"
    }

    _json_schema = {
        "title": "Example:Create",
        "type": "object",
        "required": ["name", "description"],
        "additionalProperties": False,
        "properties": {
            "name": {
                "type": "string",
                "minLength": 3,
                "maxLength": 100
            },
            "description": {
                "type": "string",
                "minLength": 10
            },
            "status": {
                "type": "string",
                "enum": ["active", "inactive"],
                "default": "active"
            }
        }
    }

    def for_json(self):
        """Convert to JSON serializable format"""
        return {
            "id": str(self._id),
            "name": self.name,
            "description": self.description,
            "status": self.status,
            "created_at": str(self.created_at) if self.created_at else None,
            "updated_at": str(self.updated_at) if self.updated_at else None,
            "user_id": str(self.user_id) if self.user_id else None
        }

3. API Endpoints (example_api.py)¶

# -*- coding: utf-8 -*-
from flask import request
from bson import ObjectId
from mongoframes import Q

from ..utils.logging import get_logger
from ..api.helpers import make_response_json
from ..api.blueprint import AppBlueprint
from ..api.errors import APIError
from ..users.auth import Auth
from .example import Example

logger = get_logger(__name__)

example_api = AppBlueprint("example", __name__)
api = example_api

@api.GET("/examples")
@api.accepts(["application/json"])
def list_examples():
    """
    List all examples with pagination and filtering

    Query Parameters:
    - page: Page number (default: 1)
    - limit: Items per page (default: 20, max: 100)
    - status: Filter by status
    - search: Search in name and description

    :return: List of examples with pagination
    :rtype: object
    """
    try:
        # Get query parameters
        page = int(request.args.get("page", 1))
        limit = min(int(request.args.get("limit", 20)), 100)
        status = request.args.get("status")
        search = request.args.get("search")

        # Build query
        query = {}
        if status:
            query["status"] = status
        if search:
            query["$or"] = [
                {"name": {"$regex": search, "$options": "i"}},
                {"description": {"$regex": search, "$options": "i"}}
            ]

        # Calculate pagination
        skip = (page - 1) * limit
        total_count = Example.count(query)
        total_pages = (total_count + limit - 1) // limit

        # Get results
        examples = Example.many(query, skip=skip, limit=limit)
        examples_data = [example.for_json() for example in examples]

        return make_response_json(
            payload_obj={
                "examples": examples_data,
                "pagination": {
                    "page": page,
                    "limit": limit,
                    "total_count": total_count,
                    "total_pages": total_pages,
                    "has_next": page < total_pages,
                    "has_prev": page > 1
                },
                "status": "success"
            },
            status_code=200
        )
    except Exception as e:
        logger.error("Error listing examples: %s", str(e))
        raise e

@api.GET("/examples/<example_id>")
@api.accepts(["application/json"])
def get_example(example_id):
    """
    Get a specific example by ID

    :param example_id: Example ObjectId
    :return: Example details
    :rtype: object
    """
    try:
        # Validate ObjectId
        if not ObjectId.is_valid(example_id):
            raise APIError.BadRequestError("Invalid example ID format")

        example = Example.by_id(ObjectId(example_id))
        if not example:
            raise APIError.NotFoundError("Example not found")

        return make_response_json(
            payload_obj={
                "example": example.for_json(),
                "status": "success"
            },
            status_code=200
        )
    except Exception as e:
        logger.error("Error getting example %s: %s", example_id, str(e))
        raise e

@api.POST("/examples")
@api.accepts(["application/json"])
@api.validate_json(Example._json_schema)
@Auth.auth
def create_example():
    """
    Create a new example

    :return: Created example details
    :rtype: object
    """
    try:
        data = request.get_json()

        # Add metadata
        data["created_at"] = arrow.Arrow.utcnow().datetime
        data["user_id"] = g.user_id

        # Create and save
        example = Example(data)
        result = example.insert()

        logger.info("Created example %s by user %s", str(result._id), g.user_id)

        return make_response_json(
            payload_obj={
                "message": "Example created successfully",
                "example": result.for_json(),
                "status": "success"
            },
            status_code=201
        )
    except Exception as e:
        logger.error("Error creating example: %s", str(e))
        raise e

@api.PUT("/examples/<example_id>")
@api.accepts(["application/json"])
@api.validate_json(Example._json_schema)
@Auth.auth
def update_example(example_id):
    """
    Update an existing example

    :param example_id: Example ObjectId
    :return: Updated example details
    :rtype: object
    """
    try:
        # Validate ObjectId
        if not ObjectId.is_valid(example_id):
            raise APIError.BadRequestError("Invalid example ID format")

        example = Example.by_id(ObjectId(example_id))
        if not example:
            raise APIError.NotFoundError("Example not found")

        # Check ownership or permissions
        if example.user_id != ObjectId(g.user_id):
            raise APIError.ForbiddenError("Cannot modify this example")

        # Update data
        data = request.get_json()
        data["updated_at"] = arrow.Arrow.utcnow().datetime

        # Apply updates
        for key, value in data.items():
            if hasattr(example, key):
                setattr(example, key, value)

        example.update()

        logger.info("Updated example %s by user %s", example_id, g.user_id)

        return make_response_json(
            payload_obj={
                "message": "Example updated successfully",
                "example": example.for_json(),
                "status": "success"
            },
            status_code=200
        )
    except Exception as e:
        logger.error("Error updating example %s: %s", example_id, str(e))
        raise e

@api.DELETE("/examples/<example_id>")
@api.accepts(["application/json"])
@Auth.auth
def delete_example(example_id):
    """
    Delete an example

    :param example_id: Example ObjectId
    :return: Deletion status
    :rtype: object
    """
    try:
        # Validate ObjectId
        if not ObjectId.is_valid(example_id):
            raise APIError.BadRequestError("Invalid example ID format")

        example = Example.by_id(ObjectId(example_id))
        if not example:
            raise APIError.NotFoundError("Example not found")

        # Check ownership or permissions
        if example.user_id != ObjectId(g.user_id):
            raise APIError.ForbiddenError("Cannot delete this example")

        # Delete the example
        example.delete()

        logger.info("Deleted example %s by user %s", example_id, g.user_id)

        return make_response_json(
            payload_obj={
                "message": "Example deleted successfully",
                "status": "success"
            },
            status_code=200
        )
    except Exception as e:
        logger.error("Error deleting example %s: %s", example_id, str(e))
        raise e

4. Service Registration (in app.py)¶

# Add to create_app() function
from .app_example.example_api import example_api
app.register_blueprint(example_api)

Summary¶

This blueprint provides a comprehensive guide for building new microservices that follow the established patterns in the AppGain Admin Server project. Key points to remember:

Follow the established directory structure and naming conventions
Use the custom AppBlueprint class for consistent API patterns
Implement proper error handling with custom error classes
Follow the MongoFrames data model pattern
Use consistent logging and monitoring patterns
Implement proper authentication and authorization
Follow the established code style and documentation standards
Test thoroughly and maintain code quality

By following this blueprint, new microservices will integrate seamlessly with the existing codebase and maintain consistency across the entire project.