Modern Web Development with Python Flask: Building Production-Ready Applications

 1# app/__init__.py
 2from flask import Flask
 3from flask_sqlalchemy import SQLAlchemy
 4from flask_migrate import Migrate
 5from flask_login import LoginManager
 6from flask_cors import CORS
 7from flask_caching import Cache
 8from flask_limiter import Limiter
 9from flask_limiter.util import get_remote_address
10import redis
11
12# Initialize extensions
13db = SQLAlchemy()
14migrate = Migrate()
15login_manager = LoginManager()
16cache = Cache()
17limiter = Limiter(key_func=get_remote_address)
18
19def create_app(config_class='app.config.ProductionConfig'):
20    app = Flask(__name__)
21    app.config.from_object(config_class)
22    
23    # Initialize extensions with app
24    db.init_app(app)
25    migrate.init_app(app, db)
26    login_manager.init_app(app)
27    CORS(app)
28    cache.init_app(app)
29    limiter.init_app(app)
30    
31    # Configure login manager
32    login_manager.login_view = 'auth.login'
33    login_manager.login_message = 'Please log in to access this page.'
34    
35    # Register blueprints
36    from app.main import bp as main_bp
37    app.register_blueprint(main_bp)
38    
39    from app.auth import bp as auth_bp
40    app.register_blueprint(auth_bp, url_prefix='/auth')
41    
42    from app.api import bp as api_bp
43    app.register_blueprint(api_bp, url_prefix='/api/v1')
44    
45    # Error handlers
46    register_error_handlers(app)
47    
48    return app
49
50def register_error_handlers(app):
51    @app.errorhandler(404)
52    def not_found_error(error):
53        return {'error': 'Resource not found'}, 404
54    
55    @app.errorhandler(500)
56    def internal_error(error):
57        db.session.rollback()
58        return {'error': 'Internal server error'}, 500
59    
60    @app.errorhandler(429)
61    def ratelimit_handler(e):
62        return {'error': 'Rate limit exceeded', 'retry_after': str(e.retry_after)}, 429

 1# app/config.py
 2import os
 3from datetime import timedelta
 4
 5class Config:
 6    SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key-change-in-production'
 7    
 8    # Database
 9    SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or 'sqlite:///app.db'
10    SQLALCHEMY_TRACK_MODIFICATIONS = False
11    SQLALCHEMY_ENGINE_OPTIONS = {
12        'pool_timeout': 20,
13        'pool_recycle': -1,
14        'pool_pre_ping': True
15    }
16    
17    # Redis
18    REDIS_URL = os.environ.get('REDIS_URL') or 'redis://localhost:6379/0'
19    
20    # Caching
21    CACHE_TYPE = 'redis'
22    CACHE_REDIS_URL = REDIS_URL
23    CACHE_DEFAULT_TIMEOUT = 300
24    
25    # Rate limiting
26    RATELIMIT_STORAGE_URL = REDIS_URL
27    RATELIMIT_DEFAULT = "1000 per hour"
28    
29    # JWT
30    JWT_SECRET_KEY = os.environ.get('JWT_SECRET_KEY') or SECRET_KEY
31    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=1)
32    JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30)
33    
34    # Email
35    MAIL_SERVER = os.environ.get('MAIL_SERVER') or 'localhost'
36    MAIL_PORT = int(os.environ.get('MAIL_PORT') or 25)
37    MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'false').lower() in ['true', 'on', '1']
38    MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
39    MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
40    
41    # File uploads
42    MAX_CONTENT_LENGTH = 16 * 1024 * 1024  # 16MB
43    UPLOAD_FOLDER = os.environ.get('UPLOAD_FOLDER') or 'uploads'
44    ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}
45
46class DevelopmentConfig(Config):
47    DEBUG = True
48    SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or 'sqlite:///app_dev.db'
49    RATELIMIT_ENABLED = False
50
51class TestingConfig(Config):
52    TESTING = True
53    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
54    WTF_CSRF_ENABLED = False
55    RATELIMIT_ENABLED = False
56
57class ProductionConfig(Config):
58    DEBUG = False
59    
60    # Enhanced security for production
61    SESSION_COOKIE_SECURE = True
62    SESSION_COOKIE_HTTPONLY = True
63    SESSION_COOKIE_SAMESITE = 'Lax'
64    PERMANENT_SESSION_LIFETIME = timedelta(hours=2)
65    
66    # Logging
67    LOG_TO_STDOUT = os.environ.get('LOG_TO_STDOUT')

  1# app/models.py
  2from datetime import datetime, timezone
  3from flask_sqlalchemy import SQLAlchemy
  4from flask_login import UserMixin
  5from werkzeug.security import generate_password_hash, check_password_hash
  6import uuid
  7
  8db = SQLAlchemy()
  9
 10class TimestampMixin:
 11    """Mixin for adding timestamp fields to models"""
 12    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), nullable=False)
 13    updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc), 
 14                          onupdate=lambda: datetime.now(timezone.utc), nullable=False)
 15
 16class User(UserMixin, db.Model, TimestampMixin):
 17    __tablename__ = 'users'
 18    
 19    id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
 20    username = db.Column(db.String(80), unique=True, nullable=False, index=True)
 21    email = db.Column(db.String(120), unique=True, nullable=False, index=True)
 22    password_hash = db.Column(db.String(255), nullable=False)
 23    is_active = db.Column(db.Boolean, default=True, nullable=False)
 24    is_admin = db.Column(db.Boolean, default=False, nullable=False)
 25    last_login = db.Column(db.DateTime)
 26    
 27    # Relationships
 28    posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
 29    comments = db.relationship('Comment', backref='author', lazy='dynamic', cascade='all, delete-orphan')
 30    
 31    def set_password(self, password):
 32        self.password_hash = generate_password_hash(password)
 33    
 34    def check_password(self, password):
 35        return check_password_hash(self.password_hash, password)
 36    
 37    def to_dict(self, include_email=False):
 38        data = {
 39            'id': self.id,
 40            'username': self.username,
 41            'is_active': self.is_active,
 42            'created_at': self.created_at.isoformat(),
 43            'last_login': self.last_login.isoformat() if self.last_login else None
 44        }
 45        if include_email:
 46            data['email'] = self.email
 47        return data
 48    
 49    def __repr__(self):
 50        return f'<User {self.username}>'
 51
 52class Post(db.Model, TimestampMixin):
 53    __tablename__ = 'posts'
 54    
 55    id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
 56    title = db.Column(db.String(200), nullable=False)
 57    content = db.Column(db.Text, nullable=False)
 58    is_published = db.Column(db.Boolean, default=False, nullable=False)
 59    view_count = db.Column(db.Integer, default=0, nullable=False)
 60    
 61    # Foreign keys
 62    author_id = db.Column(db.String(36), db.ForeignKey('users.id'), nullable=False, index=True)
 63    
 64    # Relationships
 65    comments = db.relationship('Comment', backref='post', lazy='dynamic', cascade='all, delete-orphan')
 66    tags = db.relationship('Tag', secondary='post_tags', backref='posts')
 67    
 68    @staticmethod
 69    def get_published():
 70        return Post.query.filter_by(is_published=True)
 71    
 72    def increment_view_count(self):
 73        self.view_count += 1
 74        db.session.commit()
 75    
 76    def to_dict(self, include_content=False):
 77        data = {
 78            'id': self.id,
 79            'title': self.title,
 80            'is_published': self.is_published,
 81            'view_count': self.view_count,
 82            'created_at': self.created_at.isoformat(),
 83            'author': self.author.username,
 84            'comment_count': self.comments.count(),
 85            'tags': [tag.name for tag in self.tags]
 86        }
 87        if include_content:
 88            data['content'] = self.content
 89        return data
 90
 91class Comment(db.Model, TimestampMixin):
 92    __tablename__ = 'comments'
 93    
 94    id = db.Column(db.String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
 95    content = db.Column(db.Text, nullable=False)
 96    is_approved = db.Column(db.Boolean, default=False, nullable=False)
 97    
 98    # Foreign keys
 99    author_id = db.Column(db.String(36), db.ForeignKey('users.id'), nullable=False)
100    post_id = db.Column(db.String(36), db.ForeignKey('posts.id'), nullable=False)
101    
102    def to_dict(self):
103        return {
104            'id': self.id,
105            'content': self.content,
106            'created_at': self.created_at.isoformat(),
107            'author': self.author.username,
108            'is_approved': self.is_approved
109        }
110
111class Tag(db.Model):
112    __tablename__ = 'tags'
113    
114    id = db.Column(db.Integer, primary_key=True)
115    name = db.Column(db.String(50), unique=True, nullable=False)
116    
117    def __repr__(self):
118        return f'<Tag {self.name}>'
119
120# Association table for many-to-many relationship
121post_tags = db.Table('post_tags',
122    db.Column('post_id', db.String(36), db.ForeignKey('posts.id'), primary_key=True),
123    db.Column('tag_id', db.Integer, db.ForeignKey('tags.id'), primary_key=True)
124)

  1# app/api/__init__.py
  2from flask import Blueprint
  3from flask_restx import Api
  4
  5bp = Blueprint('api', __name__)
  6api = Api(bp, doc='/docs/', title='Modern Flask API', version='1.0',
  7          description='A production-ready Flask API with authentication and rate limiting')
  8
  9from app.api import users, posts, auth
 10
 11# app/api/auth.py
 12from flask import current_app
 13from flask_restx import Namespace, Resource, fields
 14from flask_jwt_extended import create_access_token, create_refresh_token, jwt_required, get_jwt_identity
 15from app.models import User, db
 16from app.api import api
 17
 18auth_ns = api.namespace('auth', description='Authentication operations')
 19
 20# Request/Response models
 21login_model = api.model('Login', {
 22    'username': fields.String(required=True, description='Username'),
 23    'password': fields.String(required=True, description='Password')
 24})
 25
 26token_model = api.model('Token', {
 27    'access_token': fields.String(description='JWT access token'),
 28    'refresh_token': fields.String(description='JWT refresh token'),
 29    'user': fields.Nested(api.model('UserInfo', {
 30        'id': fields.String,
 31        'username': fields.String,
 32        'email': fields.String
 33    }))
 34})
 35
 36@auth_ns.route('/login')
 37class AuthLogin(Resource):
 38    @auth_ns.expect(login_model)
 39    @auth_ns.marshal_with(token_model)
 40    @auth_ns.response(200, 'Login successful')
 41    @auth_ns.response(401, 'Invalid credentials')
 42    def post(self):
 43        """Authenticate user and return JWT tokens"""
 44        data = api.payload
 45        
 46        user = User.query.filter_by(username=data['username']).first()
 47        
 48        if user and user.check_password(data['password']) and user.is_active:
 49            # Update last login
 50            user.last_login = datetime.now(timezone.utc)
 51            db.session.commit()
 52            
 53            # Create tokens
 54            access_token = create_access_token(identity=user.id)
 55            refresh_token = create_refresh_token(identity=user.id)
 56            
 57            return {
 58                'access_token': access_token,
 59                'refresh_token': refresh_token,
 60                'user': user.to_dict(include_email=True)
 61            }
 62        
 63        auth_ns.abort(401, 'Invalid credentials')
 64
 65@auth_ns.route('/refresh')
 66class AuthRefresh(Resource):
 67    @jwt_required(refresh=True)
 68    @auth_ns.marshal_with(api.model('AccessToken', {
 69        'access_token': fields.String(description='New JWT access token')
 70    }))
 71    @auth_ns.response(200, 'Token refreshed successfully')
 72    def post(self):
 73        """Refresh access token using refresh token"""
 74        user_id = get_jwt_identity()
 75        user = User.query.get(user_id)
 76        
 77        if not user or not user.is_active:
 78            auth_ns.abort(401, 'User not found or inactive')
 79        
 80        access_token = create_access_token(identity=user_id)
 81        return {'access_token': access_token}
 82
 83# app/api/posts.py
 84from flask_restx import Namespace, Resource, fields
 85from flask_jwt_extended import jwt_required, get_jwt_identity
 86from app.models import Post, User, db
 87from app.api import api
 88from app import cache, limiter
 89
 90posts_ns = api.namespace('posts', description='Blog post operations')
 91
 92post_model = api.model('Post', {
 93    'id': fields.String(readOnly=True),
 94    'title': fields.String(required=True, description='Post title'),
 95    'content': fields.String(required=True, description='Post content'),
 96    'is_published': fields.Boolean(description='Publication status'),
 97    'view_count': fields.Integer(readOnly=True),
 98    'created_at': fields.DateTime(readOnly=True),
 99    'author': fields.String(readOnly=True),
100    'tags': fields.List(fields.String, description='Post tags')
101})
102
103post_list_model = api.model('PostList', {
104    'posts': fields.List(fields.Nested(post_model)),
105    'total': fields.Integer,
106    'page': fields.Integer,
107    'per_page': fields.Integer,
108    'has_next': fields.Boolean,
109    'has_prev': fields.Boolean
110})
111
112@posts_ns.route('/')
113class PostList(Resource):
114    @posts_ns.marshal_with(post_list_model)
115    @cache.cached(timeout=300, query_string=True)
116    @limiter.limit("100 per minute")
117    def get(self):
118        """Get paginated list of published posts"""
119        page = int(request.args.get('page', 1))
120        per_page = min(int(request.args.get('per_page', 10)), 100)
121        
122        posts_query = Post.get_published().order_by(Post.created_at.desc())
123        paginated_posts = posts_query.paginate(
124            page=page, per_page=per_page, error_out=False
125        )
126        
127        return {
128            'posts': [post.to_dict() for post in paginated_posts.items],
129            'total': paginated_posts.total,
130            'page': paginated_posts.page,
131            'per_page': paginated_posts.per_page,
132            'has_next': paginated_posts.has_next,
133            'has_prev': paginated_posts.has_prev
134        }
135    
136    @posts_ns.expect(post_model)
137    @posts_ns.marshal_with(post_model, code=201)
138    @jwt_required()
139    @limiter.limit("10 per minute")
140    def post(self):
141        """Create a new post"""
142        user_id = get_jwt_identity()
143        data = api.payload
144        
145        post = Post(
146            title=data['title'],
147            content=data['content'],
148            author_id=user_id,
149            is_published=data.get('is_published', False)
150        )
151        
152        db.session.add(post)
153        db.session.commit()
154        
155        # Clear cache
156        cache.delete_memoized('get_published_posts')
157        
158        return post.to_dict(include_content=True), 201
159
160@posts_ns.route('/<post_id>')
161class PostDetail(Resource):
162    @posts_ns.marshal_with(post_model)
163    @cache.cached(timeout=600)
164    def get(self, post_id):
165        """Get a specific post by ID"""
166        post = Post.query.get_or_404(post_id)
167        
168        if not post.is_published:
169            posts_ns.abort(404, 'Post not found')
170        
171        # Increment view count asynchronously
172        post.increment_view_count()
173        
174        return post.to_dict(include_content=True)
175    
176    @posts_ns.expect(post_model)
177    @posts_ns.marshal_with(post_model)
178    @jwt_required()
179    def put(self, post_id):
180        """Update a post"""
181        user_id = get_jwt_identity()
182        post = Post.query.get_or_404(post_id)
183        
184        if post.author_id != user_id:
185            posts_ns.abort(403, 'Access denied')
186        
187        data = api.payload
188        post.title = data.get('title', post.title)
189        post.content = data.get('content', post.content)
190        post.is_published = data.get('is_published', post.is_published)
191        
192        db.session.commit()
193        
194        # Clear cache
195        cache.delete(f"post_{post_id}")
196        cache.delete_memoized('get_published_posts')
197        
198        return post.to_dict(include_content=True)