RESTful APIs with Flask

from flask_wtf.csrf import CSRFProtect

# Initialize CSRF protection
csrf = CSRFProtect()

def create_app():
    app = Flask(__name__)
    
    # Configure CSRF protection
    app.config['WTF_CSRF_ENABLED'] = True
    app.config['WTF_CSRF_SECRET_KEY'] = os.environ.get('CSRF_SECRET_KEY')
    csrf.init_app(app)
    
    # Apply CSRF protection selectively
    # Exempt certain endpoints if needed (like webhooks)
    @csrf.exempt
    @app.route('/api/webhook', methods=['POST'])
    def webhook():
        # Webhook processing without CSRF checks
        pass
        
    # For protected routes, clients need to include CSRF token
    # in X-CSRF-Token header or as a form field
    @app.route('/api/protected', methods=['POST'])
    def protected_endpoint():
        # This endpoint is protected by CSRF
        # The request must include a valid CSRF token
        return jsonify({"status": "success"})

from marshmallow import Schema, fields, validate, validates, ValidationError
import re

class UserInputSchema(Schema):
    """
    Advanced input validation schema with multiple validation layers
    
    Combines marshmallow built-in validators with custom validators
    to ensure data integrity and security
    """
    # String validation with specific pattern matching
    name = fields.String(
        required=True, 
        validate=[
            validate.Length(min=1, max=50),
            # Regex pattern to prevent injection
            validate.Regexp(r'^[a-zA-Z0-9_\- ]+$', error="Name contains invalid characters")
        ]
    )
    
    # Email validation (built-in)
    email = fields.Email(required=True)
    
    # Numeric range validation
    age = fields.Integer(validate=validate.Range(min=0, max=120))
    
    # Custom validation methods for complex rules
    password = fields.String(required=True)
    
    @validates('password')
    def validate_password(self, value):
        """
        Custom validator for password complexity requirements
        
        Enforces multiple security rules simultaneously
        """
        if len(value) < 8:
            raise ValidationError("Password must be at least 8 characters")
            
        if not re.search(r'[A-Z]', value):
            raise ValidationError("Password must contain an uppercase letter")
            
        if not re.search(r'[a-z]', value):
            raise ValidationError("Password must contain a lowercase letter")
            
        if not re.search(r'[0-9]', value):
            raise ValidationError("Password must contain a number")
            
        if not re.search(r'[^A-Za-z0-9]', value):
            raise ValidationError("Password must contain a special character")

# Usage in an API endpoint
@app.route('/api/users', methods=['POST'])
def create_user():
    try:
        # Validate and sanitize input data
        schema = UserInputSchema()
        data = schema.load(request.json)
    except ValidationError as err:
        # Return detailed validation errors
        return jsonify({"error": "Validation failed", "details": err.messages}), 422
        
    # Process validated data safely
    # ...

from flask import Flask
from flask_talisman import Talisman

app = Flask(__name__)

# Configure Talisman for advanced HTTP security headers
talisman = Talisman(
    app,
    # Content Security Policy to prevent XSS attacks
    content_security_policy={
        'default-src': "'self'",  # Default to same-origin
        'img-src': '*',           # Images can be loaded from anywhere
        'script-src': [
            "'self'", 
            "'unsafe-inline'",    # Allow inline scripts (careful with this)
            'https://cdnjs.cloudflare.com'  # Allow specific CDNs
        ],
        'style-src': ["'self'", "'unsafe-inline'"],
        'connect-src': ["'self'", "https://api.example.com"]
    },
    # Force HTTPS on all connections
    force_https=True,
    # HTTP Strict Transport Security header
    # Ensures future requests use HTTPS only
    strict_transport_security=True,
    # X-Frame-Options header to prevent clickjacking
    frame_options='DENY',
    # X-Content-Type-Options header to prevent MIME sniffing
    content_type_nosniff=True,
    # Feature-Policy header to restrict browser features
    feature_policy={
        'geolocation': "'none'",
        'microphone': "'none'",
        'camera': "'none'",
        'payment': "'self'"
    },
    # Referrer-Policy header to control referrer information
    referrer_policy='strict-origin-when-cross-origin'
)

# For API-only applications, you might want to set CORS headers
from flask_cors import CORS

# Configure CORS for API endpoints
cors = CORS(
    app, 
    resources={
        # Apply to all API routes
        r"/api/*": {
            # Define allowed origins (can be specific domains)
            "origins": [
                "https://your-frontend-app.com", 
                "https://admin.your-app.com"
            ],
            # Allow specific headers
            "allow_headers": [
                "Content-Type", 
                "Authorization", 
                "X-API-Key"
            ],
            # Allow specific HTTP methods
            "methods": ["GET", "POST", "PUT", "DELETE", "OPTIONS"],
            # Allow credentials (cookies, authorization headers)
            "supports_credentials": True,
            # Cache preflight requests
            "max_age": 86400  # 24 hours
        }
    }
)

from flask import Flask, request, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

app = Flask(__name__)

# Initialize rate limiter
limiter = Limiter(
    app,
    key_func=get_remote_address,  # Use IP address as identifier
    default_limits=["200 per day", "50 per hour"],  # Global limits
    storage_uri="redis://localhost:6379/0"  # Redis for distributed systems
)

# More sophisticated key function that includes the user ID when authenticated
def get_identifier():
    """
    Custom rate limit key function
    
    Uses authenticated user ID if available, falls back to IP address
    """
    # Try to get JWT identity if available
    try:
        from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request
        verify_jwt_in_request(optional=True)
        user_id = get_jwt_identity()
        if user_id:
            return f"user:{user_id}"
    except:
        pass
    
    # Fall back to IP address if user isn't authenticated
    return f"ip:{get_remote_address()}"

# Create limiter with custom key function
limiter = Limiter(
    app,
    key_func=get_identifier,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="redis://localhost:6379/0"
)

# Apply different limits to different endpoints
@app.route("/api/public")
@limiter.limit("1000 per day")  # More generous limit for public endpoint
def public_endpoint():
    return jsonify({"status": "public data"})

@app.route("/api/sensitive")
@limiter.limit("10 per minute")  # Stricter limit for sensitive endpoint
def sensitive_endpoint():
    return jsonify({"status": "sensitive data"})

# Dynamic rate limits based on user role
@app.route("/api/data")
@jwt_required()
def data_endpoint():
    # Get current user's role from JWT claims
    claims = get_jwt()
    user_role = claims.get("role", "user")
    
    # Apply different limits based on role
    if user_role == "admin":
        # Allow more requests for admins
        limiter.limit("1000 per hour")(lambda: None)()
    elif user_role == "premium":
        # Medium limit for premium users
        limiter.limit("100 per hour")(lambda: None)()
    else:
        # Stricter limit for regular users
        limiter.limit("20 per hour")(lambda: None)()
    
    return jsonify({"status": "data endpoint"})

# Create a custom error handler for rate limit exceeded
@app.errorhandler(429)
def ratelimit_handler(e):
    """Custom response for rate limit exceeded"""
    return jsonify({
        "error": "Rate limit exceeded",
        "message": str(e.description),
        "retry_after": e.headers.get('Retry-After', 60)
    }), 429

import logging
import json
from datetime import datetime
from flask import request, g
import uuid
from logging.handlers import RotatingFileHandler
import traceback

# Create a security logger with rotating file handler
def configure_security_logging(app):
    """Set up specialized security logging for the application"""
    
    # Configure the security logger
    security_logger = logging.getLogger('api.security')
    security_logger.setLevel(logging.INFO)
    
    # Create a rotating file handler (10MB files, max 10 files)
    handler = RotatingFileHandler(
        'logs/security.log', 
        maxBytes=10*1024*1024,  # 10MB
        backupCount=10
    )
    
    # Create formatter with detailed information
    formatter = logging.Formatter(
        '%(asctime)s - %(levelname)s - %(message)s'
    )
    handler.setFormatter(formatter)
    security_logger.addHandler(handler)
    
    # Attach the logger to the app for easy access
    app.security_logger = security_logger
    
    # Request ID middleware
    @app.before_request
    def log_request_info():
        """Generate and log request details for each incoming request"""
        
        # Generate a unique request ID and attach to request context
        request_id = str(uuid.uuid4())
        g.request_id = request_id
        
        # Get relevant security information for logging
        log_data = {
            'request_id': request_id,
            'timestamp': datetime.utcnow().isoformat(),
            'method': request.method,
            'path': request.path,
            'remote_addr': request.remote_addr,
            'headers': {
                # Include relevant headers, exclude sensitive ones
                k: v for k, v in request.headers.items()
                if k.lower() in ['user-agent', 'content-type', 'accept']
            },
            'query_params': request.args.to_dict(),
        }
        
        # Check for authentication information
        try:
            from flask_jwt_extended import get_jwt_identity, verify_jwt_in_request
            verify_jwt_in_request(optional=True)
            user_id = get_jwt_identity()
            if user_id:
                log_data['user_id'] = user_id
        except:
            pass
        
        # Log for audit trail
        app.security_logger.info(
            f"Request {request_id}: {json.dumps(log_data)}"
        )
    
    # Register error logger
    @app.errorhandler(Exception)
    def log_exception(e):
        """Log security-relevant exceptions"""
        
        # Don't log 404s as security issues
        if hasattr(e, 'code') and e.code == 404:
            return None
            
        # Create detailed error log
        error_data = {
            'request_id': getattr(g, 'request_id', 'unknown'),
            'exception': str(e),
            'exception_class': e.__class__.__name__,
            'traceback': traceback.format_exc(),
            'path': request.path,
            'method': request.method,
            'remote_addr': request.remote_addr
        }
        
        # Log at error level
        app.security_logger.error(
            f"Exception: {json.dumps(error_data)}"
        )
        
        # Re-raise for Flask's error handling
        raise e
    
    # Authentication event logger
    def log_auth_event(event_type, user_id, success, details=None):
        """Log authentication and authorization events"""
        
        auth_data = {
            'event_type': event_type,  # login, logout, access_denied, etc.
            'user_id': user_id,
            'success': success,
            'timestamp': datetime.utcnow().isoformat(),
            'remote_addr': request.remote_addr,
            'user_agent': request.headers.get('User-Agent', 'unknown'),
            'request_id': getattr(g, 'request_id', 'unknown'),
            'details': details
        }
        
        # Log auth events with appropriate level
        if success:
            app.security_logger.info(f"Auth: {json.dumps(auth_data)}")
        else:
            app.security_logger.warning(f"Auth: {json.dumps(auth_data)}")
    
    # Attach auth logger to app context
    app.log_auth_event = log_auth_event
    
    return app

# Usage example
def create_app():
    app = Flask(__name__)
    # Configure security logging
    app = configure_security_logging(app)
    # ... other configuration
    return app

# Using the auth event logger
@app.route("/api/auth/login", methods=["POST"])
def login():
    username = request.json.get("username")
    password = request.json.get("password")
    
    user = User.query.filter_by(username=username).first()
    
    if user and user.check_password(password):
        # Successful login
        app.log_auth_event(
            event_type="login", 
            user_id=user.id, 
            success=True
        )
        # Generate tokens and return response
        return jsonify({"status": "success"})
    else:
        # Failed login
        app.log_auth_event(
            event_type="login", 
            user_id=username,  # Just log the attempted username
            success=False,
            details={"reason": "Invalid credentials"}
        )
        return jsonify({"error": "Invalid credentials"}), 401

from flask import request, url_for
from flask_restful import Resource
from base64 import b64encode, b64decode
import json

class BookListAPI(Resource):
    def get(self):
        """
        List books with cursor-based pagination
        
        This implementation:
        1. Uses a cursor based on the primary key (id)
        2. Encodes the pagination state in a base64 token for clean URLs
        3. Provides next/prev page links in response for easier navigation
        4. Handles edge cases like empty results
        """
        # Parse pagination parameters
        page_size = min(int(request.args.get('page_size', 20)), 100)  # Cap at 100 items
        cursor = request.args.get('cursor')  # Cursor for pagination
        
        # Prepare query - start with all books ordered by ID
        query = Book.query.order_by(Book.id.asc())
        
        # Apply cursor filter if provided
        cursor_id = None
        if cursor:
            try:
                # Decode the cursor (base64 encoded JSON)
                cursor_data = json.loads(b64decode(cursor).decode('utf-8'))
                cursor_id = cursor_data.get('id')
                if cursor_id:
                    # Filter items after the cursor
                    query = query.filter(Book.id > cursor_id)
            except (ValueError, json.JSONDecodeError):
                # Handle invalid cursor
                return {"error": "Invalid cursor parameter"}, 400
        
        # Execute query with limit
        books = query.limit(page_size + 1).all()  # Get one extra to check if more pages exist
        
        # Check if there are more results
        has_more = len(books) > page_size
        if has_more:
            books = books[:-1]  # Remove the extra item
        
        # Build response with pagination metadata
        result = {
            'items': [book.to_dict() for book in books],
            'page_size': page_size,
            'has_more': has_more
        }
        
        # Generate next page cursor if there are more results
        if has_more and books:
            # Create cursor from the last item's ID
            last_id = books[-1].id
            next_cursor = b64encode(json.dumps({'id': last_id}).encode('utf-8')).decode('utf-8')
            result['next_cursor'] = next_cursor
            
            # Include full URL to next page for convenience
            next_url = url_for('book_list', 
                              page_size=page_size, 
                              cursor=next_cursor, 
                              _external=True)
            result['next_page'] = next_url
        
        # Include current position metadata if items exist
        if books:
            result['position'] = {
                'first_id': books[0].id,
                'last_id': books[-1].id,
                'count': len(books)
            }
            
        return result, 200

from sqlalchemy import desc

class OptimizedBookListAPI(Resource):
    def get(self):
        """
        Optimized pagination using keyset pagination (seek method)
        
        This is highly efficient for large datasets as it:
        1. Uses indexed columns for filtering
        2. Doesn't need to scan through offset rows
        3. Maintains consistency even with data modifications
        """
        # Parse parameters
        page_size = min(int(request.args.get('page_size', 20)), 100)
        sort_field = request.args.get('sort', 'created_at')  # Field to sort by
        sort_dir = request.args.get('dir', 'desc')  # Sort direction
        
        # Get the "after" values from query parameters
        after_id = request.args.get('after_id')
        after_value = request.args.get('after_value')
        
        # Base query with proper sorting
        query = Book.query
        
        # Apply sorting (ensure sort_field is valid to prevent SQL injection)
        valid_sort_fields = {'id', 'title', 'created_at', 'updated_at'}
        if sort_field not in valid_sort_fields:
            return {"error": f"Invalid sort field. Must be one of: {valid_sort_fields}"}, 400
            
        # Get the model's column for the requested sort field
        sort_column = getattr(Book, sort_field)
        
        # Apply sort direction
        if sort_dir.lower() == 'desc':
            sort_column = desc(sort_column)
            # Secondary sort by ID for stability
            query = query.order_by(sort_column, desc(Book.id))
            
            # Apply keyset pagination filter for descending order
            if after_id and after_value:
                # Convert after_value to appropriate type
                typed_after_value = self._convert_type(after_value, sort_field)
                
                # This is the key part: efficient keyset filtering
                query = query.filter(
                    # Either the sort value is less than after_value
                    (getattr(Book, sort_field) < typed_after_value) |
                    # Or the sort value is the same but the ID is less than after_id
                    ((getattr(Book, sort_field) == typed_after_value) & 
                     (Book.id < int(after_id)))
                )
        else:
            # Ascending order logic (similar but with opposite comparisons)
            query = query.order_by(sort_column, Book.id)
            
            if after_id and after_value:
                typed_after_value = self._convert_type(after_value, sort_field)
                query = query.filter(
                    (getattr(Book, sort_field) > typed_after_value) |
                    ((getattr(Book, sort_field) == typed_after_value) & 
                     (Book.id > int(after_id)))
                )
        
        # Execute the query with limit
        books = query.limit(page_size + 1).all()
        
        # Check if there are more pages
        has_more = len(books) > page_size
        if has_more:
            books = books[:-1]
            
        # Build the response
        result = {
            'items': [book.to_dict() for book in books],
            'has_more': has_more
        }
        
        # Add pagination links if there are results
        if books:
            last_book = books[-1]
            # Store the values needed for the next page
            result['after'] = {
                'id': last_book.id,
                'value': getattr(last_book, sort_field)
            }
            
            # Build next page URL
            next_url = url_for(
                'optimized_book_list',
                page_size=page_size,
                sort=sort_field,
                dir=sort_dir,
                after_id=last_book.id,
                after_value=getattr(last_book, sort_field),
                _external=True
            )
            result['next_page'] = next_url
            
        return result, 200
        
    def _convert_type(self, value, field):
        """Helper to convert string values to appropriate Python types"""
        if field == 'created_at' or field == 'updated_at':
            from datetime import datetime
            # Assume ISO format
            return datetime.fromisoformat(value)
        elif field == 'id':
            return int(value)
        else:
            # String fields need no conversion
            return value

from flask import Flask, Blueprint
from flask_restful import Api

# Create different blueprints for each API version
api_v1 = Blueprint('api_v1', __name__, url_prefix='/api/v1')
api_v2 = Blueprint('api_v2', __name__, url_prefix='/api/v2')

# Create Flask-RESTful API instances for each version
api_v1_instance = Api(api_v1)
api_v2_instance = Api(api_v2)

# Register resources with version-specific implementations
api_v1_instance.add_resource(BookListResourceV1, '/books')
api_v1_instance.add_resource(BookResourceV1, '/books/')

api_v2_instance.add_resource(BookListResourceV2, '/books')
api_v2_instance.add_resource(BookResourceV2, '/books/')

# V2 might have additional endpoints
api_v2_instance.add_resource(BookSearchResource, '/books/search')

# In your main app file
app = Flask(__name__)
app.register_blueprint(api_v1)
app.register_blueprint(api_v2)

# This allows both versions to coexist
if __name__ == '__main__':
    app.run(debug=True)

from flask import Flask, request, jsonify
from flask_restful import Resource

class BookResource(Resource):
    def get(self, book_id):
        """
        Get book details with content negotiation for versioning
        
        Clients specify version via Accept header:
        - V1: Accept: application/vnd.myapi.v1+json
        - V2: Accept: application/vnd.myapi.v2+json
        """
        # Get the Accept header
        accept_header = request.headers.get('Accept', '')
        
        # Determine version from Accept header
        if 'application/vnd.myapi.v2+json' in accept_header:
            return self.get_v2(book_id)
        else:
            # Default to v1 for backward compatibility
            return self.get_v1(book_id)
    
    def get_v1(self, book_id):
        """V1 implementation"""
        book = Book.query.get_or_404(book_id)
        # Basic representation
        return {
            'id': book.id,
            'title': book.title,
            'author': book.author,
            'year': book.publication_year
        }
    
    def get_v2(self, book_id):
        """V2 implementation with enhanced data"""
        book = Book.query.get_or_404(book_id)
        # Enhanced representation with more fields and HATEOAS links
        return {
            'id': book.id,
            'title': book.title,
            'author': {
                'name': book.author,
                'bio': book.author_bio
            },
            'publication': {
                'year': book.publication_year,
                'publisher': book.publisher,
                'isbn': book.isbn
            },
            'summary': book.summary,
            'genres': [genre.name for genre in book.genres],
            '_links': get_book_links(book_id, book)
        }

from flask import Flask, request, jsonify, make_response
import hashlib
import json

app = Flask(__name__)

@app.route('/api/books/')
def get_book(book_id):
    """
    Get book details with HTTP caching support using ETags
    
    ETags allow clients to make conditional requests, reducing
    bandwidth and processing when resource hasn't changed
    """
    # Fetch the book from database
    book = Book.query.get_or_404(book_id)
    
    # Get last modified timestamp
    last_modified = book.updated_at.timestamp()
    
    # Generate data dictionary (without ETag yet)
    data = {
        'id': book.id,
        'title': book.title,
        'author': book.author,
        'updated_at': last_modified
    }
    
    # Generate ETag (hash of the data)
    data_json = json.dumps(data, sort_keys=True)
    etag = hashlib.md5(data_json.encode()).hexdigest()
    
    # Check if client sent If-None-Match header
    if_none_match = request.headers.get('If-None-Match')
    
    if if_none_match and if_none_match == etag:
        # Resource not modified, return 304 Not Modified with empty body
        return '', 304
    
    # Resource modified or first request, return full response with ETag
    response = make_response(jsonify(data))
    response.headers['ETag'] = etag
    response.headers['Cache-Control'] = 'max-age=300'  # Cache for 5 minutes
    return response

from flask import Flask, request, jsonify
import redis
import json
import hashlib
from functools import wraps

app = Flask(__name__)
# Initialize Redis client
redis_client = redis.StrictRedis(host='localhost', port=6379, db=0)

def cached(timeout=300):
    """
    Decorator for caching API responses in Redis
    
    Parameters:
    - timeout: Cache expiration time in seconds (default: 5 minutes)
    
    This implements a pattern of:
    1. Generate a unique cache key based on endpoint and parameters
    2. Check if response exists in cache
    3. Return cached response or execute view and cache result
    """
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            # Generate a cache key based on the request
            cache_key = generate_cache_key()
            
            # Try to get cached response
            cached_response = redis_client.get(cache_key)
            if cached_response:
                # Return cached response if found
                return json.loads(cached_response)
            
            # Execute the view function if no cached response
            response = f(*args, **kwargs)
            
            # Cache the response (only if status code is 200)
            if response[1] == 200:  # Assuming response is (data, status_code)
                redis_client.setex(
                    cache_key,
                    timeout,
                    json.dumps(response[0])
                )
            
            return response
        return decorated_function
    return decorator

def generate_cache_key():
    """Generate a cache key based on the request"""
    # Include path, query parameters, and possibly auth info in key
    key_parts = [
        request.path,
        str(sorted(request.args.items())),
    ]
    
    # Add user info if authenticated request (for user-specific caching)
    # This ensures users don't see each other's data from cache
    user_id = get_current_user_id()  # Your authentication logic
    if user_id:
        key_parts.append(str(user_id))
    
    # Create a hash of the parts
    key = hashlib.md5('|'.join(key_parts).encode()).hexdigest()
    return f"api:cache:{key}"

@app.route('/api/books')
@cached(timeout=300)  # Cache for 5 minutes
def get_books():
    """Get a list of books with caching"""
    # Apply filters from query params
    query = Book.query
    
    # Apply filters from request args
    author = request.args.get('author')
    if author:
        query = query.filter(Book.author.like(f'%{author}%'))
    
    # This expensive database query now only runs on cache misses
    books = query.all()
    return jsonify([book.to_dict() for book in books]), 200

@app.route('/api/cache/invalidate', methods=['POST'])
def invalidate_cache():
    """Invalidate cache for specific patterns (admin only)"""
    # Check if user is admin
    if not is_admin():
        return jsonify({"error": "Unauthorized"}), 403
        
    pattern = request.json.get('pattern', 'api:cache:*')
    
    # Find keys matching pattern
    keys = redis_client.keys(pattern)
    if keys:
        # Use redis pipeline for atomic operation with multiple keys
        pipeline = redis_client.pipeline()
        for key in keys:
            pipeline.delete(key)
        pipeline.execute()
        return jsonify({"message": f"Invalidated {len(keys)} cache entries"}), 200
    else:
        return jsonify({"message": "No matching cache entries found"}), 200

from flask import Flask
from flask_smorest import Api, Blueprint, abort
from marshmallow import Schema, fields

app = Flask(__name__)

app.config["API_TITLE"] = "Books API"
app.config["API_VERSION"] = "v1"
app.config["OPENAPI_VERSION"] = "3.0.2"
app.config["OPENAPI_URL_PREFIX"] = "/"
app.config["OPENAPI_SWAGGER_UI_PATH"] = "/swagger-ui"
app.config["OPENAPI_SWAGGER_UI_URL"] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"

api = Api(app)

# Define schemas for request/response validation and documentation
class BookSchema(Schema):
    id = fields.Int(dump_only=True)
    title = fields.Str(required=True)
    author = fields.Str(required=True)
    publication_year = fields.Int()
    isbn = fields.Str()

# Create a blueprint with automatic documentation
books_blp = Blueprint(
    "Books", "books", url_prefix="/api/books",
    description="Operations on books"
)

@books_blp.route("/")
class BookList:
    @books_blp.response(200, BookSchema(many=True))
    @books_blp.doc(description="Get all books")
    def get(self):
        """List all books"""
        return Book.query.all()
    
    @books_blp.arguments(BookSchema)
    @books_blp.response(201, BookSchema)
    @books_blp.doc(description="Create a new book")
    def post(self, book_data):
        """Create a new book"""
        book = Book(**book_data)
        db.session.add(book)
        db.session.commit()
        return book

@books_blp.route("/")
class BookResource:
    @books_blp.response(200, BookSchema)
    @books_blp.alt_response(404, description="Book not found")
    @books_blp.doc(description="Get a specific book")
    def get(self, book_id):
        """Get a book by ID"""
        book = Book.query.get_or_404(book_id)
        return book

# Register blueprint with the API
api.register_blueprint(books_blp)

import pytest
import json
from app import create_app
from app.models import db as _db, Book

@pytest.fixture
def app():
    """Create application for the tests."""
    app = create_app('testing')
    app.config['TESTING'] = True
    return app

@pytest.fixture
def db(app):
    """Create a database for the tests."""
    with app.app_context():
        _db.create_all()
        yield _db
        _db.session.close()
        _db.drop_all()

@pytest.fixture
def client(app):
    """A test client for the app."""
    return app.test_client()

@pytest.fixture
def book(db):
    """Create a test book."""
    book = Book(
        title="Test Book",
        author="Test Author",
        publication_year=2020,
        isbn="1234567890"
    )
    db.session.add(book)
    db.session.commit()
    return book

def test_get_books(client):
    """Test getting all books."""
    response = client.get('/api/books')
    assert response.status_code == 200
    assert isinstance(json.loads(response.data), list)

def test_get_book(client, book):
    """Test getting a specific book."""
    response = client.get(f'/api/books/{book.id}')
    assert response.status_code == 200
    data = json.loads(response.data)
    assert data['title'] == "Test Book"
    assert data['author'] == "Test Author"

def test_create_book(client, db):
    """Test creating a new book."""
    book_data = {
        'title': 'New Book',
        'author': 'New Author',
        'publication_year': 2023
    }
    response = client.post(
        '/api/books',
        data=json.dumps(book_data),
        content_type='application/json'
    )
    assert response.status_code == 201
    data = json.loads(response.data)
    assert data['title'] == book_data['title']
    
    # Verify the book was actually created in the database
    book = Book.query.filter_by(title='New Book').first()
    assert book is not None
    assert book.author == 'New Author'

# Dockerfile for Flask API
FROM python:3.10-slim

WORKDIR /app

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

# Copy application code
COPY . .

# Set environment variables
ENV FLASK_APP=app.py
ENV FLASK_ENV=production
ENV PYTHONUNBUFFERED=1

# Run gunicorn with 4 workers
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "wsgi:app"]

# docker-compose.yml
version: '3'

services:
  api:
    build: .
    restart: always
    ports:
      - "5000:5000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/database
      - SECRET_KEY=${SECRET_KEY}
      - JWT_SECRET_KEY=${JWT_SECRET_KEY}
    depends_on:
      - db
      - redis
    networks:
      - app-network
      
  db:
    image: postgres:13
    volumes:
      - postgres_data:/var/lib/postgresql/data/
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=database
    networks:
      - app-network
      
  redis:
    image: redis:alpine
    volumes:
      - redis_data:/data
    networks:
      - app-network
      
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx/conf.d:/etc/nginx/conf.d
      - ./nginx/ssl:/etc/nginx/ssl
    depends_on:
      - api
    networks:
      - app-network

networks:
  app-network:

volumes:
  postgres_data:
  redis_data: