GuardianFlow

A production-ready Django GraphQL API with enterprise-grade features including authentication, rate limiting, background tasks, and comprehensive error handling.

Table of Contents

• Quick Start
• Architecture
• Project Structure
• Configuration
• API Documentation
• Development Guide
• Deployment

---

Quick Start

Prerequisites

• Python 3.11+
• Redis (for caching and Celery results)
• RabbitMQ (for Celery message broker)

Installation

# Clone repository
git clone https://github.com/yourusername/guardianflow.git
cd guardianflow

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

# Start Redis and RabbitMQ (Docker)
docker run -d -p 6379:6379 redis
docker run -d -p 5672:5672 rabbitmq

# Run migrations
python manage.py migrate

# Create superuser
python manage.py createsuperuser

# Start development server
python manage.py runserver

Start Background Workers

# Terminal 2: Celery worker
celery -A guardianflow worker --loglevel=info

# Terminal 3: Celery beat (scheduled tasks)
celery -A guardianflow beat --loglevel=info

Access Points

Endpoint	Description
http://localhost:8000/graphql/	GraphQL API + GraphiQL IDE
http://localhost:8000/admin/	Django Admin
http://localhost:8000/health/	Health check
http://localhost:8000/health/ready/	Detailed readiness check

---

Architecture

System Overview

┌─────────────────────────────────────────────────────────────────────────┐
│                              CLIENTS                                     │
│                    (Web, Mobile, Third-party Services)                   │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                                    ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                         DJANGO APPLICATION                               │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                    HTTP Middleware Layer                         │    │
│  │  • RequestLoggingMiddleware - Logs all HTTP requests            │    │
│  │  • RateLimitMiddleware - Global rate limiting (200 req/min)     │    │
│  └─────────────────────────────────────────────────────────────────┘    │
│                                    │                                     │
│                                    ▼                                     │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                    GraphQL Layer (Graphene)                      │    │
│  │  ┌───────────────────────────────────────────────────────────┐  │    │
│  │  │              GraphQL Middleware                            │  │    │
│  │  │  • LoggingMiddleware - Logs operations                    │  │    │
│  │  │  • RateLimitMiddleware - Per-operation limits             │  │    │
│  │  └───────────────────────────────────────────────────────────┘  │    │
│  │                              │                                   │    │
│  │                              ▼                                   │    │
│  │  ┌───────────────────────────────────────────────────────────┐  │    │
│  │  │                   Schema (Queries/Mutations)               │  │    │
│  │  │  • AccountQueries - User queries                          │  │    │
│  │  │  • AccountMutations - User mutations                      │  │    │
│  │  └───────────────────────────────────────────────────────────┘  │    │
│  └─────────────────────────────────────────────────────────────────┘    │
│                                    │                                     │
│                                    ▼                                     │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                      Service Layer                               │    │
│  │  • AccountService - Business logic for user operations          │    │
│  │  • Validators - Input validation                                 │    │
│  │  • Exceptions - Domain-specific errors                          │    │
│  └─────────────────────────────────────────────────────────────────┘    │
│                                    │                                     │
│                                    ▼                                     │
│  ┌─────────────────────────────────────────────────────────────────┐    │
│  │                      Data Layer                                  │    │
│  │  • Models - Django ORM models                                   │    │
│  │  • Serializers - Data validation & transformation               │    │
│  └─────────────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────────────┘
         │                              │                      │
         ▼                              ▼                      ▼
    ┌─────────┐                  ┌───────────┐          ┌───────────┐
    │ SQLite/ │                  │   Redis   │          │ RabbitMQ  │
    │ Postgres│                  │  (Cache)  │          │ (Broker)  │
    └─────────┘                  └───────────┘          └───────────┘
                                       │                      │
                                       └──────────┬───────────┘
                                                  ▼
                                       ┌───────────────────┐
                                       │   Celery Workers  │
                                       │  (Background Jobs)│
                                       └───────────────────┘

Request Flow

1. HTTP Request arrives
   ↓
2. Django Middleware processes (logging, rate limiting)
   ↓
3. GraphQL View receives request
   ↓
4. GraphQL Middleware processes (logging, operation rate limiting)
   ↓
5. Schema routes to appropriate Query/Mutation
   ↓
6. Resolver calls Service Layer
   ↓
7. Service Layer validates, processes, interacts with Models
   ↓
8. Response returns through the stack

---

Project Structure

guardianflow/
├── manage.py                    # Django management script
├── requirements.txt             # Python dependencies
├── db.sqlite3                   # SQLite database (dev only)
│
├── logs/                        # Application logs
│   └── guardianflow.log
│
└── guardianflow/                # Main Django project
    ├── __init__.py              # Celery app initialization
    ├── settings.py              # Django settings
    ├── urls.py                  # URL routing
    ├── celery.py                # Celery configuration
    ├── wsgi.py                  # WSGI entry point
    ├── asgi.py                  # ASGI entry point
    │
    ├── account/                 # Account app (users)
    │   ├── models.py            # User model
    │   ├── service.py           # Business logic
    │   ├── serializers.py       # Data serialization
    │   ├── validators.py        # Input validation
    │   ├── exceptions.py        # Custom exceptions
    │   ├── tasks.py             # Celery tasks
    │   └── admin.py             # Admin configuration
    │
    ├── core/                    # Core utilities
    │   ├── ratelimit.py         # Rate limiter utilities
    │   └── middleware/          # Django HTTP middleware
    │       ├── __init__.py
    │       ├── rate_limit.py    # HTTP rate limiting
    │       └── request_logging.py
    │
    ├── graphql/                 # GraphQL API
    │   ├── api.py               # Main schema
    │   ├── pagination.py        # Shared pagination
    │   │
    │   ├── middleware/          # GraphQL middleware
    │   │   ├── __init__.py
    │   │   ├── logging.py       # Operation logging
    │   │   └── rate_limit.py    # Operation rate limiting
    │   │
    │   └── account/             # Account GraphQL module
    │       ├── schema.py        # Queries & Mutations
    │       ├── types.py         # GraphQL types
    │       ├── filters.py       # Query filters
    │       └── mutations/       # Mutation handlers
    │           └── account_create.py
    │
    └── health/                  # Health check endpoints
        ├── __init__.py
        ├── urls.py
        └── views.py

---

Configuration

Environment Variables

Create a .env file in the project root:

# Django
DEBUG=True
SECRET_KEY=your-secret-key-here
ALLOWED_HOSTS=localhost,127.0.0.1

# Database (for production)
DATABASE_URL=postgres://user:pass@localhost:5432/guardianflow

# Redis
REDIS_URL=redis://localhost:6379/1

# Celery
CELERY_BROKER_URL=amqp://guest:guest@localhost:5672//
CELERY_RESULT_BACKEND=redis://localhost:6379/0

# Sentry (optional)
SENTRY_DSN=https://your-sentry-dsn
SENTRY_ENVIRONMENT=development
SENTRY_TRACES_SAMPLE_RATE=0.1

Rate Limiting

Configure in settings.py:

RATELIMIT_SETTINGS = {
    # Global HTTP rate limit
    'GLOBAL_REQUESTS': 200,      # requests
    'GLOBAL_WINDOW': 60,         # seconds
    
    # GraphQL operation limits
    'QUERY_REQUESTS': 100,
    'QUERY_WINDOW': 60,
    
    'MUTATION_REQUESTS': 30,
    'MUTATION_WINDOW': 60,
    
    'INTROSPECTION_REQUESTS': 10,
    'INTROSPECTION_WINDOW': 60,
}

Celery Beat Schedule

Periodic tasks in settings.py:

CELERY_BEAT_SCHEDULE = {
    'cleanup-inactive-users': {
        'task': 'guardianflow.account.tasks.cleanup_inactive_users',
        'schedule': 60 * 60 * 24,  # Daily
        'args': (90,),             # 90 days inactive
    },
    'generate-daily-report': {
        'task': 'guardianflow.account.tasks.generate_user_activity_report',
        'schedule': 60 * 60 * 24,  # Daily
    },
}

---

API Documentation

GraphQL Endpoint

URL: POST /graphql/

Headers:

Content-Type: application/json

Queries

Get Current User

query {
  me {
    id
    email
    isActive
    isAdmin
    dateOfBirth
  }
}

List Users (with pagination & filtering)

query {
  users(
    first: 10
    after: "cursor"
    email_Icontains: "example"
    isActive: true
    orderBy: "-date_joined"
  ) {
    totalCount
    edgeCount
    pageInfo {
      hasNextPage
      hasPreviousPage
      startCursor
      endCursor
    }
    edges {
      cursor
      node {
        id
        email
        isActive
        isAdmin
        dateOfBirth
      }
    }
  }
}

Get Single User

query {
  user(id: "QWNjb3VudFR5cGU6MQ==") {
    id
    email
    isActive
  }
}

Mutations

Create User

mutation {
  accountCreate(input: {
    email: "user@example.com"
    password: "securepassword123"
    dateOfBirth: "1990-01-15"
  }) {
    success
    message
    user {
      id
      email
    }
    errors {
      field
      messages
    }
  }
}

Error Responses

All errors follow this structure:

{
  "errors": [
    {
      "message": "Error description",
      "path": ["fieldName"],
      "extensions": {
        "code": "ERROR_CODE"
      }
    }
  ]
}

Error Codes:

Code	Description
VALIDATION_ERROR	Input validation failed
INVALID_CREDENTIALS	Authentication failed
EMAIL_ALREADY_EXISTS	Email already registered
RATE_LIMIT_EXCEEDED	Too many requests
NOT_FOUND	Resource not found

---

Development Guide

Code Style

We follow PEP 8 with these tools:

• Black: Code formatting
• isort: Import sorting
• flake8: Linting

# Format code
black guardianflow/
isort guardianflow/

# Check linting
flake8 guardianflow/

Adding a New App

1. Create the app structure:

mkdir -p guardianflow/newapp

2. Create required files:

guardianflow/newapp/
├── __init__.py
├── models.py        # Database models
├── service.py       # Business logic
├── serializers.py   # Data validation
├── validators.py    # Input validators
├── exceptions.py    # Custom exceptions
├── tasks.py         # Celery tasks
└── admin.py         # Admin registration

3. Add to INSTALLED_APPS in settings.py:

INSTALLED_APPS = [
    ...
    'guardianflow.newapp',
]

4. Create GraphQL module:

guardianflow/graphql/newapp/
├── __init__.py
├── schema.py        # Queries & Mutations
├── types.py         # GraphQL types
├── filters.py       # Query filters
└── mutations/
    └── __init__.py

5. Register in main schema (graphql/api.py):

from .newapp.schema import NewAppQueries, NewAppMutations

class Query(AccountQueries, NewAppQueries, graphene.ObjectType):
    pass

class Mutation(AccountMutations, NewAppMutations, graphene.ObjectType):
    pass

Adding a New Celery Task

1. Define in tasks.py:

from celery import shared_task

@shared_task(bind=True, max_retries=3)
def my_task(self, arg1, arg2):
    try:
        # Task logic here
        return {'status': 'success'}
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)

2. Call from service:

from .tasks import my_task

# Async execution
my_task.delay(arg1, arg2)

# With options
my_task.apply_async(
    args=[arg1, arg2],
    countdown=10,  # Delay 10 seconds
    expires=3600,  # Expire after 1 hour
)

Service Layer Pattern

All business logic should be in service classes:

# service.py
class MyService:
    @staticmethod
    def create_item(data: dict) -> Item:
        """
        Create a new item.
        
        Args:
            data: Item data dictionary
            
        Returns:
            Created Item instance
            
        Raises:
            ValidationError: If validation fails
        """
        # Validate
        validate_item_data(data)
        
        # Create
        serializer = ItemSerializer(data=data)
        if not serializer.is_valid():
            raise ValidationError(serializer.errors)
        
        item = serializer.save()
        
        # Trigger async tasks
        send_notification.delay(item.id)
        
        return item

Exception Handling

Use domain-specific exceptions:

# exceptions.py
class AppException(Exception):
    """Base exception for the app."""
    message = "An error occurred"
    code = "APP_ERROR"

class ValidationError(AppException):
    message = "Validation failed"
    code = "VALIDATION_ERROR"
    
    def __init__(self, errors: dict):
        self.errors = errors
        super().__init__(self.message)

Testing

# Run all tests
pytest

# Run with coverage
pytest --cov=guardianflow --cov-report=html

# Run specific test file
pytest guardianflow/account/tests/test_service.py

# Run specific test
pytest -k "test_create_user"

---

Deployment

Production Checklist

• Set DEBUG=False
• Generate secure SECRET_KEY
• Configure ALLOWED_HOSTS
• Set up PostgreSQL database
• Configure Redis for production
• Set up Sentry for error tracking
• Configure proper logging
• Set up HTTPS
• Run collectstatic

Docker Deployment

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

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

COPY . .

RUN python manage.py collectstatic --noinput

CMD ["gunicorn", "guardianflow.wsgi:application", "--bind", "0.0.0.0:8000"]

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgres://postgres:postgres@db:5432/guardianflow
      - REDIS_URL=redis://redis:6379/1
      - CELERY_BROKER_URL=amqp://rabbitmq:5672//
    depends_on:
      - db
      - redis
      - rabbitmq

  celery:
    build: .
    command: celery -A guardianflow worker --loglevel=info
    depends_on:
      - web
      - rabbitmq
      - redis

  celery-beat:
    build: .
    command: celery -A guardianflow beat --loglevel=info
    depends_on:
      - celery

  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=guardianflow
      - POSTGRES_PASSWORD=postgres
    volumes:
      - postgres_data:/var/lib/postgresql/data

  redis:
    image: redis:7
    volumes:
      - redis_data:/data

  rabbitmq:
    image: rabbitmq:3

volumes:
  postgres_data:
  redis_data:

Health Checks

Endpoint	Purpose	Used By
/health/	Liveness probe	Kubernetes
/health/ready/	Readiness probe	Load balancer
/health/db/	Database check	Monitoring
/health/celery/	Worker check	Monitoring
/health/cache/	Cache check	Monitoring

---

License

MIT License - see LICENSE for details.