stllrent-bootstrap 1.3.0

Comprehensive Python module designed to standardize and accelerate the development of responsive Flask APIs that leverage Celery for asynchronous workload processing

stllrent-bootstrap

🌟 Project Overview

stllrent-bootstrap is a comprehensive Python module designed to standardize and accelerate the development of responsive Flask APIs that leverage Celery for asynchronous workload processing. This module is optimized for containerized production environments and is designed to work seamlessly with industry-standard technologies like Kubernetes, RabbitMQ, and PostgreSQL.

With this module, you can focus on your API's business logic, while stllrent-bootstrap handles the robust configuration and best practices for asynchronous processing.

---

💡 Why Use stllrent-bootstrap?

In a microservices architecture, managing asynchronous communication, result persistence, and workflow monitoring can be challenging. stllrent-bootstrap addresses these challenges by providing a solid, standardized foundation, ensuring:

• Standardization and Reusability: Offers a consistent structure for initializing Flask applications and Celery Workers, promoting code reuse and simplified maintenance across multiple projects.
• Robust Asynchronous Processing: Configures Celery with high-reliability patterns, including:
  • acks_late=True: Ensures messages are acknowledged in RabbitMQ only after successful processing, preventing message loss in case of Worker failures.
  • DLQs (Dead-Letter Queues): Automatically routes messages from failed tasks to a Dead-Letter Queue in RabbitMQ, allowing for auditing, debugging, and reprocessing without data loss.
  • Extended Result Persistence: Saves detailed task results (success, failure, tracebacks) to PostgreSQL via result_extended=True.
• Cloud and Container-Ready: Pre-configured to work seamlessly with databases and message brokers as a service on any cloud provider, or in a self-hosted infrastructure. The design is optimized for deployment in Kubernetes clusters.
• Clear Code Structure: Separates responsibilities between the Flask API (task producer) and Celery Workers (task consumers), facilitating development and maintenance.
• Application Factories: Provides factory functions to create configured instances of Flask and Celery applications, simplifying initialization in your integration projects.

---

🚀 Key Features

• Centralized Configuration: Manages all environment configurations (broker URLs, backend URLs, DBs) via pydantic-settings.
• Robust Connection Management: Includes hooks to ensure proper database session closing in both Flask and Celery contexts.
• Dynamic Model Discovery: Automatically loads SQLAlchemy models defined in your project, simplifying database initialization.
• Custom Base Task: Provides a BootstrapTask class that enhances Celery tasks with improved logging, exception handling for DLQ, and automatic integration with workflow monitoring services.

---

📐 Architecture Overview

The stllrent-bootstrap module provides three main integration points:

graph TD
    A["📊 Your Flask Application<br/>(src/app.py)"]
    B["🔌 HTTP Endpoints<br/>(src/route/api.py)"]
    C["📤 Celery Task Producer<br/>(configure_celery_for_flask)"]
    D["🐰 RabbitMQ Broker<br/>(Message Queue)"]
    E["📦 PostgreSQL<br/>(Result Backend)"]
    F["⚙️ Celery Worker Pool<br/>(src/worker/*.py)"]
    G["🚀 BootstrapTask Base<br/>(Logging, Monitoring, Notifications)"]
    H["📈 Workflow Monitoring Service<br/>(FLOWMON)"]
    I["📧 Email Notifications<br/>(Retry/Reject Alerts)"]
    
    A -->|HTTP Request| B
    B -->|Queue Task| C
    C -->|Send Message| D
    D -->|Consume Task| F
    D -->|Persist Result| E
    F -->|Inherits| G
    G -->|Update Status| H
    G -->|Notify On Event| I
    E -->|Query Results| H
    
    style A fill:#4A90E2,color:#fff
    style B fill:#50C878,color:#fff
    style C fill:#FFB84D,color:#fff
    style D fill:#E74C3C,color:#fff
    style E fill:#9B59B6,color:#fff
    style F fill:#1ABC9C,color:#fff
    style G fill:#F39C12,color:#fff
    style H fill:#34495E,color:#fff
    style I fill:#E67E22,color:#fff

Flow:

1. HTTP request arrives at Flask endpoint
2. Endpoint queues a Celery task via RabbitMQ broker
3. Task message is published to RabbitMQ
4. Worker pool picks up message and executes task
5. Result and status are persisted to PostgreSQL
6. Workflow monitoring service receives status updates
7. On retry/reject, notifications are sent via email (optional)

For the full source code and additional diagrams, see docs/architecture-diagrams.md.

---

🏁 Getting Started: Integrating stllrent-bootstrap

To integrate stllrent-bootstrap into your project, it is recommended to follow a standard structure and configuration pattern to fully leverage the module's automation features. The stellrent-contract project serves as an excellent implementation example of this pattern.

1. Installation

Add stllrent-bootstrap to your project's requirements.txt file.

# Example of installing from a private Git repository
pip install stllrent-bootstrap

2. Project Structure Requirements

For stllrent-bootstrap to work correctly, your project must follow a minimal structure. The bootstrap module expects to find the following files and directories:

your_project/
├── src/
│   ├── app.py                  # Flask application entry point
│   ├── celery_app.py           # Celery App instantiation
│   ├── config/
│   │   ├── base_settings.py    # Application settings (inherits from BaseAppSettings)
│   │   └── celery_settings.py  # Celery settings (inherits from BaseCelerySettings)
│   ├── model/                  # Package for SQLAlchemy models
│   │   └── __init__.py
│   ├── route/
│   │   └── api.py              # Must contain the register_blueprints(app) function
│   └── worker/
│       └── ...                 # Modules containing your Celery tasks
└── .env

3. Configuration

Environment Variables (.env)

Your .env file must contain all the necessary environment variables. The bootstrap will load them to populate the configuration objects.

# Application Settings (Flask)
DATABASE_HOST=localhost
DATABASE_USER=user
DATABASE_PASS=password
DATABASE_PORT=5432
DATABASE_NAME=mydatabase
API_PRIMARY_PATH=api/v1
APP_NAME=my-service
APP_LOG_LEVEL=DEBUG
FLASK_ENV=development

# Celery Settings
CELERY_BROKER_URL="amqp://user:password@rabbitmq-host:5672//"
RESULT_BACKEND_URL=localhost
RESULT_BACKEND_PORT=55432
RESULT_BACKEND_USER=user
RESULT_BACKEND_PASS=password
RESULT_BACKEND_DATABASE=celery_results

Configuration Classes

In your project, create classes that inherit from the base settings classes provided by stllrent-bootstrap to extend and customize the behavior.

Example - src/config/base_settings.py:

from stllrent_bootstrap.flask.app_settings import BaseAppSettings

class Settings(BaseAppSettings):
    # Override or add service-specific settings
    APP_NAME: str = "my-contract-service"
    MODEL_DISCOVERY_PATHS: list[str] = ["model"] # Tells bootstrap where to find your models

Example - src/config/celery_settings.py:

from stllrent_bootstrap.celery.config.settings import BaseCelerySettings
from stllrent_bootstrap.celery.model.brokers import RabbitMQModel
from .base_settings import settings # Import your app settings

# Define the broker models for your queues
example_broker = RabbitMQModel(queue="my-service.my-queue.example")

class CelerySettings(BaseCelerySettings):
    APP_NAME: str = settings.APP_NAME
    celery_task_default_queue: str = example_broker.queue
    broker_models: list[RabbitMQModel] = [example_broker]

4. Application Initialization

Use the factory functions from the bootstrap to instantiate your Flask and Celery applications.

src/app.py:

from config.base_settings import Settings
from stllrent_bootstrap.flask.app_factory import create_app

settings = Settings()
app = create_app(settings)

# Optional: for local execution
if __name__ == '__main__':
    from waitress import serve
    serve(app, host="0.0.0.0", port=settings.APP_PORT)

src/celery_app.py:

from config.celery_settings import celery_settings
from stllrent_bootstrap.celery.app import create_celery_app

# List of modules where your tasks are defined for autodiscovery
PROJECT_TASK_PATHS = [
    'worker.my_worker_1',
    'worker.my_worker_2',
]

celery_app = create_celery_app(
    celery_settings=celery_settings,
    autodiscover_paths=PROJECT_TASK_PATHS
)

5. Route Definitions

The bootstrap's app_factory expects to find a route.api module with a register_blueprints function to load your application's routes.

src/route/api.py:

def register_blueprints(app):
    from route.endpoint.public import public_blueprint
    app.register_blueprint(public_blueprint)

    # Register other blueprints here

6. Creating Celery Tasks

Define your tasks in src/worker/ modules:

src/worker/payment_tasks.py:

from stllrent_bootstrap.celery.base_task import BootstrapTask
from celery_app import celery_app
from service.payment_service import process_payment
from celery.exceptions import Reject

@celery_app.task(bind=True, base=BootstrapTask, name='worker.process_payment')
def process_payment_async(self, customer_id: str, amount: float):
    """
    Async task to process a customer payment.
    
    Attributes:
        notify_on_retry: If set, sends email on retry. Default: False
        notify_on_reject: If set, sends email on reject. Default: False
        max_retries: Maximum retry attempts. Default: 3
    """
    # Optional: Enable notifications
    self.notify_on_retry = '[e-mail]'
    self.notify_on_reject = '[e-mail]'
    
    try:
        result = process_payment(customer_id, amount)
        return {'customer_id': customer_id, 'amount': amount, 'status': 'completed'}
    except TemporaryError as e:
        # Retry after 60 seconds
        raise self.retry(exc=e, countdown=60)
    except PaymentRejected as e:
        # Reject message and send to DLQ
        raise Reject(str(e), requeue=False)

7. Dispatching Tasks from Flask Endpoints

src/route/endpoint/payment.py:

from flask import Blueprint, request, jsonify
from worker.payment_tasks import process_payment_async
from stellrent_response.json_response import JsonResponse

payment_blueprint = Blueprint('payment', __name__)

@payment_blueprint.post('/payments')
def create_payment():
    body = request.get_json()
    
    # Validate
    if not body.get('customer_id') or not body.get('amount'):
        return JsonResponse.error('Missing fields', 400)
    
    # Queue async task
    task = process_payment_async.apply_async(
        args=[body['customer_id'], body['amount']],
        countdown=5  # Execute after 5 seconds
    )
    
    return JsonResponse.success(
        data={'request_id': task.id, 'status': 'queued'},
        status_code=202
    )

8. Optional: Email Notifications for Retry & Reject

To enable email notifications when tasks retry or are rejected, add to .env:

# .env
NOTIFICATION_EMAIL_RELAY=smtp.gmail.com
NOTIFICATION_EMAIL_FROM=[e-mail]
NOTIFICATION_EMAIL_STARTTLS=True

Then in your task, set the notification recipients:

@celery_app.task(bind=True, base=BootstrapTask)
def my_task(self):
    self.notify_on_retry = '[e-mail]'
    self.notify_on_reject = '[e-mail]'
    # ... rest of code

---

📊 Logging & Monitoring

The module standardizes task events with structured log codes for observability:

Task Event Codes

Event	Code	Level	When
Task Scheduled (Retry)	TASK_RETRY_SCHEDULED	ERROR	Task failed, scheduled for retry
Task Rejected (Requeued)	TASK_REJECTED_REQUEUED	ERROR	Task rejected, message returns to queue
Task Rejected (No Requeue)	TASK_REJECTED_NO_REQUEUE	ERROR	Task rejected, message goes to DLQ
Task Failure (Unhandled)	TASK_FAILURE_UNHANDLED	CRITICAL	Unexpected exception in task
Task Failure	TASK_FAILURE	ERROR	Task entered FAILURE state
Monitoring Update	TASK_MONITORING_UPDATE	INFO	Starting status update to monitoring service
Monitoring Updated	TASK_MONITORING_UPDATED	DEBUG	Status successfully updated
Notification Sending	TASK_NOTIFICATION_SENDING	DEBUG	Sending email notification
Notification Sent	TASK_NOTIFICATION_SENT	DEBUG	Email sent successfully
Notification Error	TASK_NOTIFICATION_ERROR	ERROR	Failed to send notification

Example Log Entry:

[TASK_RETRY_SCHEDULED] [task=abc-123] [retries=1] [max_retries=3] [reason=ConnectionError(...)]
[TASK_NOTIFICATION_SENDING] [type=retry] [task=abc-123] [recipient=[e-mail]]

Task Execution Flow & Event Handling

graph TD
    A["Task Dispatched<br/>(HTTP Endpoint)"]
    B["Task Queued<br/>(RabbitMQ)"]
    C["Worker Processes"]
    
    D["✅ Success<br/>(TASK_SUCCESS)"]
    E["⏱️ Temporary Failure<br/>(TASK_RETRY_SCHEDULED)"]
    F["🚫 Permanent Failure<br/>(TASK_REJECTED_NO_REQUEUE)"]
    G["♻️ Requeue<br/>(TASK_REJECTED_REQUEUED)"]
    H["❌ Unhandled Exception<br/>(TASK_FAILURE_UNHANDLED)"]
    
    I["Update Monitoring<br/>(TASK_MONITORING_UPDATE)"]
    J["Send Email Alert<br/>(TASK_NOTIFICATION_SENDING)"]
    K["Log Event<br/>(Structured Log)"]
    L["Persist Result<br/>(PostgreSQL)"]
    
    A --> B
    B --> C
    
    C -->|No Error| D
    C -->|Retry Exception| E
    C -->|Reject - No Requeue| F
    C -->|Reject - Requeue| G
    C -->|Unhandled Error| H
    
    D --> I
    E --> J
    E --> I
    F --> J
    F --> I
    G --> B
    H --> I
    
    D --> K
    E --> K
    F --> K
    G --> K
    H --> K
    
    I --> L
    J --> L
    
    style A fill:#4A90E2,color:#fff
    style B fill:#E74C3C,color:#fff
    style C fill:#F39C12,color:#fff
    style D fill:#50C878,color:#fff
    style E fill:#FFB84D,color:#fff
    style F fill:#E67E22,color:#fff
    style G fill:#9B59B6,color:#fff
    style H fill:#C0392B,color:#fff
    style I fill:#34495E,color:#fff
    style J fill:#1ABC9C,color:#fff
    style K fill:#2980B9,color:#fff
    style L fill:#8E44AD,color:#fff

Monitoring Integration

All tasks automatically report their status to the configured workflow monitoring service. Set the environment variables:

FLOWMON_URL=https://flowmon.mycompany.com
FLOWMON_API_PRIMARY_PATH=workflow

---

✅ Testing Tasks

Use the fixtures provided by bootstrap for testing:

src/tests/test_payment_tasks.py:

import pytest
from celery.exceptions import Reject
from worker.payment_tasks import process_payment_async

@pytest.mark.celery
def test_payment_task_success(celery_app_test):
    result = process_payment_async.apply_async(
        args=['cust-123', 100.0],
        app=celery_app_test
    )
    assert result.successful()
    assert result.result['status'] == 'completed'

@pytest.mark.celery
def test_payment_task_retry_on_temporary_error(celery_app_test):
    with pytest.raises(Retry):
        process_payment_async.apply_async(
            args=['cust-invalid', 0.0],
            app=celery_app_test
        ).get()

@pytest.mark.celery
def test_payment_task_reject(celery_app_test):
    with pytest.raises(Reject):
        process_payment_async.apply_async(
            args=['cust-fraud', 999999.0],
            app=celery_app_test
        ).get()

Run tests:

cd src && pytest -v --cov

---

🔧 Troubleshooting

Common Issues

"ModuleNotFoundError: No module named 'config.base_settings'"

Cause: The bootstrap expects config.base_settings module in your project.

Fix: Ensure your project structure matches:

src/
├── config/
│   ├── __init__.py
│   ├── base_settings.py    ← Must exist
│   └── celery_settings.py  ← Must exist
└── ...

"ValidationError: CELERY_BROKER_URL - Field required"

Cause: Missing required environment variables.

Fix: Ensure .env has all required variables or set them in your shell:

export CELERY_BROKER_URL="amqp://guest:guest@localhost:5672//"
export RESULT_BACKEND_URL="localhost"
# ... other vars

Tasks are not being processed

Cause: Worker not running or not subscribed to correct queues.

Fix: Start the worker explicitly:

cd src && celery -A celery_app worker -l info

Monitoring updates are failing

Cause: FLOWMON_URL not set or unreachable.

Fix: Check environment and network connectivity:

echo $FLOWMON_URL
curl -v $FLOWMON_URL/health

OAuth2 token lock contention (high concurrency)

Symptom: Logs show repeated [OAUTH_TOKEN_LOCK_TIMEOUT] and [OAUTH_TOKEN_FALLBACK].

Cause: Multiple workers competing for the distributed lock faster than one can refresh.

Fix:

1. Increase OAUTH_CACHE_RETRY_ATTEMPTS (default 3) to allow more wait cycles.
2. Review lock_ttl_seconds — ensure it is longer than the token refresh round-trip.
3. Check ElastiCache latency; add a CloudWatch alarm for EngineCPUUtilization.

Redis unavailable (ElastiCache unreachable)

Symptom: Logs show [OAUTH_TOKEN_CACHE_ERROR] but service continues to work.

Cause: Redis connection failure. The bootstrap falls back transparently to direct OAuth2 fetch.

Fix:

1. Confirm OAUTH_CACHE_REDIS_URL is correct and the cluster is healthy.
2. Check Kubernetes network policies and Security Groups on the ElastiCache cluster.
3. Verify OAUTH_CACHE_TLS_ENABLED=True matches the cluster's TLS configuration.

# Test Redis connectivity from inside the pod
python -c "import redis; redis.Redis.from_url('$OAUTH_CACHE_REDIS_URL').ping()"

Bearer token expiring before TTL

Symptom: Downstream services return 401 Unauthorized before the cache key expires.

Cause: Token server returns a shorter expires_in than expected, or the buffer is too small.

Fix: Increase the per-alias buffer so the cached TTL is safely within the actual expiry:

# Add 90 s of buffer for the SERVICE1 alias (default is 60 s via OAUTH_CACHE_TTL_BUFFER_SECONDS)
OAUTH_SERVICE1_TTL_BUFFER_SECONDS=90

---

🔑 OAuth2 Shared Token Cache (Redis)

Available since v1.2.0. Disabled by default — existing integrations are unaffected.

Architecture

graph TD
    A["Worker / Flask Request"] -->|"get_bearer_token()"| B{"OAUTH_TOKEN_CACHE_ENABLED?"}
    B -->|"False (default)"| C["Direct OAuth2 fetch"]
    B -->|"True"| D{"Redis Cache hit?"}
    D -->|"Hit"| E["Return cached token"]
    D -->|"Miss"| F{"Acquire distributed lock"}
    F -->|"Lock acquired"| G["Refresh from OAuth2 endpoint"]
    G --> H["Persist token in Redis (TTL)"]
    H --> E
    F -->|"Lock not acquired (retry x3)"| I{"Cache populated by other worker?"}
    I -->|"Yes"| E
    I -->|"No after 3 attempts"| J["Direct fallback OAuth2 fetch"]
    J --> K["Return token (no cache write)"]

    style A fill:#4A90E2,color:#fff
    style C fill:#50C878,color:#fff
    style E fill:#50C878,color:#fff
    style J fill:#FFB84D,color:#fff
    style K fill:#FFB84D,color:#fff

Enabling the feature

# 1. Enable the feature flag (default: False)
OAUTH_TOKEN_CACHE_ENABLED=true

# 2. Point to your ElastiCache Redis cluster (TLS endpoint)
OAUTH_CACHE_BACKEND=redis
OAUTH_CACHE_REDIS_URL=rediss://<cluster-endpoint>:6379/0
OAUTH_CACHE_TLS_ENABLED=true

# 3. Per-alias OAuth2 credentials (one set per external service)
# SERVICE1
OAUTH_SERVICE1_TOKEN_URL=https://auth.service1.example/oauth2/token
OAUTH_SERVICE1_CLIENT_ID=<service1-client-id>
OAUTH_SERVICE1_CLIENT_SECRET=<service1-client-secret>
OAUTH_SERVICE1_SCOPE=service1.read

# SERVICE2
OAUTH_SERVICE2_TOKEN_URL=https://auth.service2.example/oauth2/token
OAUTH_SERVICE2_CLIENT_ID=<service2-client-id>
OAUTH_SERVICE2_CLIENT_SECRET=<service2-client-secret>
OAUTH_SERVICE2_SCOPE=service2.read

Security note: never commit secrets to source control.
Use AWS Secrets Manager, Vault, or Kubernetes Secrets to inject the values above.

Usage in tasks (Celery context)

from stllrent_bootstrap.auth import get_bearer_token

@celery_app.task(bind=True, base=BootstrapTask)
def sync_service1_data(self):
    token = get_bearer_token(
        app_alias="SERVICE1",
        client_id=settings.OAUTH_SERVICE1_CLIENT_ID,
        settings=settings,
    )
    headers = {"Authorization": f"Bearer {token}"}
    # ... rest of integration call


@celery_app.task(bind=True, base=BootstrapTask)
def sync_service2_data(self):
    token = get_bearer_token(
        app_alias="SERVICE2",
        client_id=settings.OAUTH_SERVICE2_CLIENT_ID,
        settings=settings,
    )
    headers = {"Authorization": f"Bearer {token}"}
    # ... rest of integration call

Usage in endpoints (Flask context)

from stllrent_bootstrap.auth import get_bearer_token_flask

@api.route("/service1/resources")
def list_service1_resources():
    token = get_bearer_token_flask(
        app_alias="SERVICE1",
        client_id=current_app.config["OAUTH_SERVICE1_CLIENT_ID"],
    )
    headers = {"Authorization": f"Bearer {token}"}
    # ... rest of integration call


@api.route("/service2/resources")
def list_service2_resources():
    token = get_bearer_token_flask(
        app_alias="SERVICE2",
        client_id=current_app.config["OAUTH_SERVICE2_CLIENT_ID"],
    )
    headers = {"Authorization": f"Bearer {token}"}
    # ... rest of integration call

Cache key convention

Tokens are stored under the key APP_ALIAS:CLIENT_ID (e.g. SERVICE1:my-client-id, SERVICE2:my-client-id). Distributed lock uses key lock:APP_ALIAS:CLIENT_ID.

Observability events

Event	Level	Meaning
OAUTH_TOKEN_CACHE_HIT	INFO	Token served from Redis — no network call
OAUTH_TOKEN_CACHE_MISS	INFO	Key absent or expired; refresh initiated
OAUTH_TOKEN_LOCK_ACQUIRED	INFO	This instance will perform the refresh
OAUTH_TOKEN_REFRESHED	INFO	Token refreshed and stored with TTL
OAUTH_TOKEN_LOCK_TIMEOUT	WARN	All retry attempts exhausted without acquiring lock
OAUTH_TOKEN_FALLBACK	WARN	Falling back to direct OAuth2 fetch (no cache write)
OAUTH_TOKEN_CACHE_ERROR	ERROR	Redis operation failed; service continues
OAUTH_TOKEN_FETCH_ERROR	ERROR	OAuth2 endpoint unreachable or returned error

Rollout guide

Phase 1 — deploy with flag disabled (default) Deploy the new version. No behaviour change. All existing integrations continue to use direct token fetch.

Phase 2 — enable per service Set OAUTH_TOKEN_CACHE_ENABLED=true for one service at a time via environment variable (ConfigMap / Kubernetes Secret). Monitor logs for OAUTH_TOKEN_CACHE_HIT and OAUTH_TOKEN_CACHE_ERROR.

Phase 3 — full rollout Once stable, enable for all services and remove the OAUTH_TOKEN_CACHE_ENABLED=false override.

Rollback Set OAUTH_TOKEN_CACHE_ENABLED=false (or remove the variable). The cache layer is bypassed immediately — no deployment required.

---

Datetime Parser Utility (BlueFleet)

Available since v1.2.0 to parse BlueFleet datetime values into UTC-0 aware datetime objects.

Supported input formats:

• %Y-%m-%dT%H:%M:%S.%fZ (example: 2026-01-19T16:59:40.787Z)
• %Y-%m-%dT%H:%M:%SZ (example: 2027-01-19T16:59:40Z)

Usage:

from stllrent_bootstrap.utils import parse_datetime

event_at = parse_datetime("2026-01-19T16:59:40.787Z")

Behavior:

• Returns a timezone-aware datetime with UTC-0 (tzinfo=timezone.utc) when parsing succeeds.
• Propagates the parser exception when parsing fails (typically ValueError).

---

📦 Requirements

• Python: >= 3.13
• Flask: >= 3.1
• Celery: >= 5.5
• SQLAlchemy: >= 2.0
• Pydantic: >= 2.11
• RabbitMQ: >= 3.8 (for production)
• PostgreSQL: >= 12 (for result backend)

Check pyproject.toml for the full dependency list.

---

🔐 Security Best Practices

1. Never commit .env file

   echo ".env" >> .gitignore
   

2. Use secrets management for production

   # Instead of .env, use environment variables from:
   # - AWS Secrets Manager
   # - HashiCorp Vault
   # - Kubernetes Secrets
   

3. Rotate credentials regularly

   • RabbitMQ user passwords
   • Database credentials
   • Email relay passwords

4. Restrict email notifications

   • Only send notifications to trusted recipients
   • Use separate email accounts for different environments (dev, prod)

5. Database connection pooling

   • Configure SQLALCHEMY_POOL_SIZE and SQLALCHEMY_MAX_OVERFLOW appropriately
   • Monitor connection exhaustion in production

6. OAuth2 client secrets

   • Never set OAUTH_<ALIAS>_CLIENT_SECRET in .env committed to source control
   • Inject via AWS Secrets Manager, HashiCorp Vault, or Kubernetes Secrets
   • Rotate client secrets periodically and update the corresponding Kubernetes Secret / Vault path

7. ElastiCache Redis security

   • Always enable TLS in transit: OAUTH_CACHE_TLS_ENABLED=true and use rediss:// URL scheme
   • Enable encryption at rest on the ElastiCache cluster (Terraform: at_rest_encryption_enabled = true)
   • Restrict Redis Security Group to only the EKS node security group
   • Disable AUTH token is not recommended for production; use IAM-based auth where available

---

📋 Quick Reference

Environment Variables Required

Variable	Default	Required	Example
DATABASE_HOST	-	✅	localhost
DATABASE_USER	-	✅	postgres
DATABASE_PASS	-	✅	secret123
DATABASE_PORT	5432	✅	5432
DATABASE_NAME	-	✅	mydb
API_PRIMARY_PATH	-	✅	api/v1
APP_NAME	-	✅	my-service
CELERY_BROKER_URL	-	✅	amqp://guest:guest@localhost:5672//
RESULT_BACKEND_URL	-	✅	localhost
RESULT_BACKEND_PORT	5432	✅	5432
RESULT_BACKEND_USER	-	✅	celery_user
RESULT_BACKEND_PASS	-	✅	secret123
RESULT_BACKEND_DATABASE	-	✅	celery_results
FLOWMON_URL	-	✅	https://flowmon.local
FLOWMON_API_PRIMARY_PATH	-	✅	workflow
NOTIFICATION_EMAIL_RELAY	-	❌	smtp.gmail.com
NOTIFICATION_EMAIL_FROM	-	❌	[e-mail]
NOTIFICATION_EMAIL_STARTTLS	True	❌	True
OAUTH_TOKEN_CACHE_ENABLED	False	❌	true
OAUTH_CACHE_BACKEND	none	❌	redis
OAUTH_CACHE_REDIS_URL	-	❌	rediss://cache.internal:6379/0
OAUTH_CACHE_TLS_ENABLED	True	❌	True
OAUTH_CACHE_TTL_BUFFER_SECONDS	60	❌	60
OAUTH_CACHE_RETRY_ATTEMPTS	3	❌	3
OAUTH_<ALIAS>_TOKEN_URL	-	❌	https://auth.example/oauth2/token
OAUTH_<ALIAS>_CLIENT_ID	-	❌	my-client-id
OAUTH_<ALIAS>_CLIENT_SECRET	-	❌	(use secrets manager)
OAUTH_<ALIAS>_SCOPE	-	❌	vehicles.read
OAUTH_<ALIAS>_TTL_BUFFER_SECONDS	60	❌	90
APP_LOG_LEVEL	INFO	❌	DEBUG
FLASK_ENV	production	❌	development

Common Commands

# Start Flask app (development)
cd src && python app.py

# Start Celery worker
cd src && celery -A celery_app worker -l info

# Monitor Celery tasks (requires Flower)
cd src && celery -A celery_app flower

# Run tests
cd src && pytest -v --cov

# Database migrations (if using Alembic)
cd src && alembic upgrade head

---

By following these steps, your project will be correctly integrated with stllrent-bootstrap, allowing you to leverage all its features for standardization and development acceleration in any containerized environment.