Create a configuration file to store environment variables by running the following command:

touch .env.development

Define your environment variables in the .env.development file to configure the application settings:

DEEPSEEK_API_KEY=your_api_key_here
FLASK_ENV=development
LOG_LEVEL=DEBUG

Update instance/config.py with the following configuration settings to manage environment variables, security policies, rate limits, logging, and API defaults:

import os
from dotenv import load_dotenv

# Load environment variables first
load_dotenv(f'.env.{os.getenv("FLASK_ENV", "production")}')

class Config:
    """Base configuration"""
    DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY")
    FLASK_ENV = os.getenv("FLASK_ENV", "production")
    
    # Security
    TALISMAN_CONFIG = {
        'content_security_policy': {
            'default-src': "'self'",
            'script-src': ["'self'", "https://trusted-cdn.com"]
        },
        'force_https': False  # Default to False, override in ProductionConfig
    }
    
    # Rate limiting
    RATE_LIMITS = ["200/hour", "50/minute"]
    
    # Logging
    LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
    LOG_FILE = "logs/prod.log"
    
    # API
    API_VERSION = "/api/v1"
    DEFAULT_MODEL = "deepseek-chat"

class DevelopmentConfig(Config):
    """Development-specific configuration"""
    DEBUG = True
    LOG_LEVEL = "DEBUG"
    LOG_FILE = "logs/dev.log"
    TEMPLATES_AUTO_RELOAD = True

class ProductionConfig(Config):
    """Production configuration"""
    TALISMAN_CONFIG = {
        'content_security_policy': {
            'default-src': "'self'",
            'script-src': ["'self'", "https://trusted-cdn.com"]
        },
        'force_https': True,
        'strict_transport_security': True
    }


class TestingConfig(Config):
    RATE_LIMITS = ["5 per minute"]  # Exact test limit
    TALISMAN_CONFIG = {
        'force_https': False,
        'strict_transport_security': True,
        'content_security_policy': {'default-src': "'self'"}
    }

This code defines a configuration system for a Flask application, handling environment-specific settings, security policies, rate limits, logging, and API defaults.

1. Load Environment Variables
   The script imports os and load_dotenv from dotenv to manage environment variables dynamically. It loads the .env file based on the FLASK_ENV variable, defaulting to "production" if not set. This allows the application to use different configurations for development, production, and testing.
2. Define the Base Configuration (Config Class)
   • The Config class acts as the foundation for all configurations.
   • It retrieves the DEEPSEEK_API_KEY and FLASK_ENV from the environment.
   • The TALISMAN_CONFIG dictionary enforces security settings, specifying a Content Security Policy (CSP) that limits sources for scripts and other content. By default, it does not enforce HTTPS.
   • The RATE_LIMITS setting controls request throttling, allowing a maximum of 200 requests per hour or 50 per minute.
   • Logging settings define the log level (default "INFO") and log file location (logs/prod.log).
   • The API version ("/api/v1") and default AI model ("deepseek-chat") are also configured.
3. Define Environment-Specific Configurations
   • DevelopmentConfig: Enables debugging (DEBUG = True), sets "DEBUG" as the log level, and reloads templates automatically.
   • ProductionConfig:
     • Overrides TALISMAN_CONFIG to enforce HTTPS (force_https=True) and strict transport security.
     • This ensures that the application always runs securely in production.
   • TestingConfig:
     • Adjusts rate limits to 5 requests per minute, mimicking real-world API throttling.
     • Uses a minimal CSP and does not enforce HTTPS for testing flexibility.

Key Takeaways:

• The code dynamically loads environment-specific settings.
• It enforces security policies using Flask-Talisman.
• It applies rate limiting to prevent API abuse.
• It configures structured logging for better debugging and monitoring.

Build the backend

Create Flask Application

Update the app/__init__.py file to initialize your Flask application with the following code snippet.

from flask import Flask
from dotenv import load_dotenv
from instance.config import (  # Add this import
    Config, 
    DevelopmentConfig, 
    ProductionConfig, 
    TestingConfig
)
from .utils.logger import configure_logging
from .security import init_security
from .routes import init_routes
import os

def create_app():
    # Load environment variables first
    env = os.getenv("FLASK_ENV", "production").lower()
    load_dotenv(f'.env.{env}')
    
    # Create app instance
    app = Flask(__name__)
    
    # Load configuration based on environment
    if env == 'development':
        app.config.from_object(DevelopmentConfig)
    elif env == 'testing':
        app.config.from_object(TestingConfig)
    else:
        app.config.from_object(ProductionConfig)
    
    # Initialize components
    configure_logging()
    init_security(app)
    init_routes(app)
    
    return app

This code initializes a Flask application with environment-based configurations, security features, and logging.

1. Loads Environment Variables – It detects the FLASK_ENV variable (defaulting to production) and loads the corresponding .env file.
2. Creates the Flask App – It initializes a Flask application instance.
3. Applies Configuration – It selects the appropriate configuration class (DevelopmentConfig, TestingConfig, or ProductionConfig).
4. Initializes Components – It sets up logging, security measures, and API routes.

Finally, the create_app function returns the configured Flask application.

Chatbot service

Update app/services/chatbot_service.py with the following code snippet to handle user queries using the DeepSeek API:

import uuid
from datetime import datetime
from typing import Dict, Optional
import structlog
from deepseek import DeepSeekAPI  # Using the provided class
import os

logger = structlog.get_logger()

class ChatbotService:
    """Handles AI-powered customer support interactions"""
    
    def __init__(self):
        self.api_key = os.getenv("DEEPSEEK_API_KEY")
        if not self.api_key:
            logger.critical("config_error", message="Missing API key")
            raise ValueError("Deepseek API key not configured")
        
        self.client = DeepSeekAPI(api_key=self.api_key)
        logger.info("service_initialized")

    def generate_response(self, user_query: str, user_ip: Optional[str] = None) -> Dict[str, any]:
        """Process customer queries with comprehensive tracking"""
        response_id = str(uuid.uuid4())
        start_time = datetime.utcnow()
        metadata = {
            "response_id": response_id,
            "timestamp": start_time.isoformat(),
            "client_ip": user_ip,
            "assistant_version": "GreenGrocer AI 1.4.2",
            "processing_time": 0.0,
            "error_code": None,
            "support_contact": "support@greengrocer.com"
        }

        try:
            if not user_query.strip():
                metadata["error_code"] = "EMPTY_QUERY_001"
                logger.error(
                    "api_failure",
                    error="Please provide a valid question",
                    query=user_query,
                    client_ip=user_ip,
                    stack_trace=self._safe_get_stacktrace(Exception("Empty query")),
                    error_code="EMPTY_QUERY_001",
                    support_contact="support@greengrocer.com",
                    response_id=str(uuid.uuid4()),
                    timestamp=datetime.utcnow().isoformat(),
                    assistant_version="GreenGrocer AI 1.4.2",
                    processing_time=0.0
                )

                return {
                    "content": "Please provide a valid question",
                    "metadata": metadata
                }

            # Handle known FAQs
            query = user_query.lower()
            if "delivery hours" in query:
                metadata["processing_time"] = self._calculate_processing_time(start_time)
                return {
                    "content": "Delivery available Mon-Sat 8:00 AM - 8:00 PM EST. "
                              "Same-day cutoff: 12:00 PM EST.",
                    "metadata": metadata
                }

            if "order status" in query:
                metadata["processing_time"] = self._calculate_processing_time(start_time)
                return {
                    "content": "Your order is en route with estimated arrival by 3:00 PM EST. "
                              "Track your order at https://track.greengrocer.com",
                    "metadata": metadata
                }

            # AI-generated response
            messages = [
                {
                    "role": "system",
                    "content": "You are a customer support agent for GreenGrocer Foods. "
                              "Respond professionally with EST timezone references. "
                              "Keep answers under 500 characters."
                },
                {
                    "role": "user",
                    "content": user_query
                }
            ]

            ai_response = self.client.chat_completion(
                prompt=messages,
                model="deepseek-chat",
                temperature=0.7,
                max_tokens=500,
                stream=False
            )

            metadata["processing_time"] = self._calculate_processing_time(start_time)
            logger.info(
                "api_success",
                query=user_query,
                response_length=len(ai_response),
                **metadata
            )

            return {
                "content": ai_response,
                "metadata": metadata
            }

        except Exception as e:
            metadata.update({
                "processing_time": self._calculate_processing_time(start_time),
                "error_code": self._get_error_code(e),
                "stack_trace": self._safe_get_stacktrace(e)
            })
            
            logger.error(
                "api_failure",
                error=str(e),
                query=user_query,
                **metadata
            )

            return {
                "content": self._get_user_friendly_error(e),
                "metadata": metadata
            }

    def _calculate_processing_time(self, start_time: datetime) -> float:
        return round((datetime.utcnow() - start_time).total_seconds(), 3)

    def _get_error_code(self, error: Exception) -> str:
        error_msg = str(error)
        if "HTTP Error 429" in error_msg:
            return "RATE_LIMIT_429"
        if "HTTP Error 5" in error_msg:
            return "SERVER_ERROR_500"
        return "CLIENT_ERROR_400"

    def _get_user_friendly_error(self, error: Exception) -> str:
        error_code = self._get_error_code(error)
        return {
            "RATE_LIMIT_429": "Our systems are busy. Please try again in a minute.",
            "SERVER_ERROR_500": "We're experiencing technical difficulties. Our team has been notified.",
            "CLIENT_ERROR_400": "Invalid request. Please check your input."
        }.get(error_code, "An unexpected error occurred. Please contact support.")

    def _safe_get_stacktrace(self, error: Exception) -> str:
        try:
            return str(error.__traceback__)
        except Exception:
            return "Stacktrace unavailable for security reasons"

This ChatbotService class provides AI-powered customer support by processing user queries, integrating structured logging, and handling errors gracefully. Here’s a breakdown of its functionality:

1. Initialization and Setup

• The __init__ method loads the DeepSeek API key from environment variables.
• If the API key is missing, it logs a critical error and raises a ValueError, preventing the chatbot from running without proper configuration.
• Once initialized, the class creates a DeepSeekAPI client and logs the successful setup.

2. Processing User Queries

The generate_response method follows a structured approach:

• It assigns a unique response ID and timestamps the request.
• It stores metadata like client IP, processing time, and error codes for tracking and debugging.
• If the query is empty, it logs an error and returns a warning message.

3. Handling Common FAQs

• If the user asks about delivery hours, it provides a predefined response:
  “Delivery available Mon-Sat 8:00 AM – 8:00 PM EST. Same-day cutoff: 12:00 PM EST.”
• If the user asks about order status, it returns a tracking link and an estimated delivery time.
• These quick responses avoid unnecessary API calls, improving efficiency.

4. AI-Generated Responses

• For general queries, the chatbot constructs a message history, including a system prompt that ensures responses remain professional, brief, and timezone-specific.
• It sends the request to DeepSeek’s AI model with parameters like temperature=0.7 (to control randomness) and max_tokens=500 (to limit response length).
• Once the AI generates a response, the chatbot logs the interaction, including the query, response length, and processing time.

5. Error Handling and Logging

• The chatbot uses structured logging (structlog) to capture both successful responses and errors.
• If an error occurs (e.g., API rate limits, server errors, or client mistakes), the bot:
  • Extracts an error code (RATE_LIMIT_429, SERVER_ERROR_500, etc.).
  • Logs debugging details, including a stack trace if available.
  • Returns a user-friendly error message, such as:
    “Our systems are busy. Please try again in a minute.”

6. Utility Methods for Reliability

Several helper methods ensure reliable execution:

• _calculate_processing_time(): Measures response time for performance monitoring.
• _get_error_code(): Categorizes errors into rate limits, server issues, or client errors.
• _get_user_friendly_error(): Maps error codes to clear messages for the user.
• _safe_get_stacktrace(): Prevents exposing sensitive error details by sanitizing stack traces.

Summary

The ChatbotService class combines AI, structured logging, and error handling to deliver fast, reliable customer support. It prioritizes FAQs, generates AI responses when needed, and logs every request for debugging and analytics.

Set up security

To enhance security, create a security folder inside the app directory and include an __init__.py file. This setup applies secure HTTP headers and rate limiting to protect your Flask application.

Update app/security/__init__.py with the following code snippet:

from flask_talisman import Talisman
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address  # Use built-in IP detection

def init_security(app):
    """Enterprise security configuration"""

    talisman_config = app.config.get('TALISMAN_CONFIG', {})

    
    # Initialize Talisman first
    Talisman(
        app,
        content_security_policy=app.config['TALISMAN_CONFIG'].get('content_security_policy'),
        force_https=app.config['TALISMAN_CONFIG'].get('force_https', False),
        strict_transport_security=app.config['TALISMAN_CONFIG'].get('strict_transport_security', True),
        frame_options='DENY'
    )

    
    # Initialize Limiter with proper configuration
    limiter = Limiter(
        app=app,
        key_func=get_remote_address,
        default_limits=app.config["RATE_LIMITS"],
        headers_enabled=True,
        storage_uri="memory://",
        strategy="moving-window"
    )
    
    # Apply rate limits to all routes by default
    limiter.init_app(app)

This code configures security features for a Flask application by enforcing secure HTTP headers and rate limiting.

How it works

1. Imports Security Modules
   • It imports Talisman to enforce security headers.
   • It imports Limiter to control request rates and prevent abuse.
   • It uses get_remote_address to identify a client’s IP for rate limiting.
2. Initializes Security in init_security(app)
   • It retrieves security settings from app.config['TALISMAN_CONFIG'].
3. Applies Security Headers Using Flask-Talisman
   • It enforces HTTPS if enabled in the configuration.
   • It applies a Content Security Policy (CSP) to prevent cross-site scripting (XSS).
   • It denies embedding via iframes (frame_options='DENY').
4. Implements Rate Limiting with Flask-Limiter
   • It restricts the number of requests per user using get_remote_address.
   • It enables rate limiting with app.config["RATE_LIMITS"].
   • It stores limits in memory and applies a moving window strategy.
5. Activates Rate Limiting
   • It calls limiter.init_app(app) to enforce the limits globally.

This setup hardens security by blocking unauthorized access, reducing bot traffic, and protecting against DDoS attacks.