chore: import upstream snapshot with attribution
Component Security Validation / Security Audit (push) Has been cancelled
Deploy to Cloudflare Pages / deploy (push) Has been cancelled

This commit is contained in:
wehub-resource-sync
2026-07-13 12:38:58 +08:00
commit bb5c75ce05
8824 changed files with 1946442 additions and 0 deletions
@@ -0,0 +1,384 @@
# Flask Application Factory Pattern
Create a scalable Flask application using the factory pattern with blueprints and configuration management.
## Purpose
This command helps you set up a Flask application using the application factory pattern, which is the recommended approach for larger Flask applications.
## Usage
```
/app-factory
```
## What this command does
1. **Creates application factory** with proper structure
2. **Sets up configuration management** for different environments
3. **Implements blueprints** for modular design
4. **Configures extensions** (database, auth, etc.)
5. **Adds error handling** and logging
## Example Output
```python
# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_mail import Mail
from flask_wtf.csrf import CSRFProtect
from flask_cors import CORS
import logging
from logging.handlers import RotatingFileHandler
import os
# Initialize extensions
db = SQLAlchemy()
migrate = Migrate()
login = LoginManager()
mail = Mail()
csrf = CSRFProtect()
cors = CORS()
def create_app(config_class=None):
"""Application factory function."""
app = Flask(__name__)
# Load configuration
if config_class is None:
config_class = os.environ.get('FLASK_CONFIG', 'development')
if isinstance(config_class, str):
from app.config import config
app.config.from_object(config[config_class])
else:
app.config.from_object(config_class)
# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
login.init_app(app)
mail.init_app(app)
csrf.init_app(app)
cors.init_app(app)
# Configure login manager
login.login_view = 'auth.login'
login.login_message = 'Please log in to access this page.'
login.login_message_category = 'info'
# Register blueprints
from app.main import bp as main_bp
app.register_blueprint(main_bp)
from app.auth import bp as auth_bp
app.register_blueprint(auth_bp, url_prefix='/auth')
from app.api import bp as api_bp
app.register_blueprint(api_bp, url_prefix='/api')
from app.admin import bp as admin_bp
app.register_blueprint(admin_bp, url_prefix='/admin')
# Error handlers
from app.errors import bp as errors_bp
app.register_blueprint(errors_bp)
# Configure logging
if not app.debug and not app.testing:
if not os.path.exists('logs'):
os.mkdir('logs')
file_handler = RotatingFileHandler(
'logs/app.log',
maxBytes=10240,
backupCount=10
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Flask application startup')
return app
# Import models (avoid circular imports)
from app import models
```
```python
# app/config.py
import os
from dotenv import load_dotenv
basedir = os.path.abspath(os.path.dirname(__file__))
load_dotenv(os.path.join(basedir, '.env'))
class Config:
"""Base configuration class."""
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///' + os.path.join(basedir, 'app.db')
SQLALCHEMY_TRACK_MODIFICATIONS = False
# Mail configuration
MAIL_SERVER = os.environ.get('MAIL_SERVER')
MAIL_PORT = int(os.environ.get('MAIL_PORT') or 587)
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1']
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
ADMIN_EMAIL = os.environ.get('ADMIN_EMAIL')
# Pagination
POSTS_PER_PAGE = 10
USERS_PER_PAGE = 50
# Upload configuration
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB
UPLOAD_FOLDER = os.path.join(basedir, 'uploads')
@staticmethod
def init_app(app):
pass
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///' + os.path.join(basedir, 'app-dev.db')
class TestingConfig(Config):
"""Testing configuration."""
TESTING = True
SQLALCHEMY_DATABASE_URI = os.environ.get('TEST_DATABASE_URL') or 'sqlite://'
WTF_CSRF_ENABLED = False
class ProductionConfig(Config):
"""Production configuration."""
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///' + os.path.join(basedir, 'app.db')
@classmethod
def init_app(cls, app):
Config.init_app(app)
# Log to stderr
import logging
from logging import StreamHandler
file_handler = StreamHandler()
file_handler.setLevel(logging.INFO)
app.logger.addHandler(file_handler)
config = {
'development': DevelopmentConfig,
'testing': TestingConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
```
```python
# app/main/__init__.py
from flask import Blueprint
bp = Blueprint('main', __name__)
from app.main import routes
```
```python
# app/main/routes.py
from flask import render_template, request, current_app
from app.main import bp
from app.models import Post
@bp.route('/')
@bp.route('/index')
def index():
"""Home page."""
page = request.args.get('page', 1, type=int)
posts = Post.query.filter_by(published=True).order_by(
Post.created_at.desc()
).paginate(
page=page,
per_page=current_app.config['POSTS_PER_PAGE'],
error_out=False
)
return render_template('index.html', posts=posts)
@bp.route('/about')
def about():
"""About page."""
return render_template('about.html')
```
```python
# app/auth/__init__.py
from flask import Blueprint
bp = Blueprint('auth', __name__)
from app.auth import routes
```
```python
# app/auth/routes.py
from flask import render_template, redirect, url_for, flash, request
from flask_login import login_user, logout_user, login_required, current_user
from werkzeug.urls import url_parse
from app import db
from app.auth import bp
from app.auth.forms import LoginForm, RegistrationForm
from app.models import User
@bp.route('/login', methods=['GET', 'POST'])
def login():
"""User login."""
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(username=form.username.data).first()
if user is None or not user.check_password(form.password.data):
flash('Invalid username or password', 'error')
return redirect(url_for('auth.login'))
login_user(user, remember=form.remember_me.data)
next_page = request.args.get('next')
if not next_page or url_parse(next_page).netloc != '':
next_page = url_for('main.index')
return redirect(next_page)
return render_template('auth/login.html', form=form)
@bp.route('/logout')
@login_required
def logout():
"""User logout."""
logout_user()
flash('You have been logged out.', 'info')
return redirect(url_for('main.index'))
@bp.route('/register', methods=['GET', 'POST'])
def register():
"""User registration."""
if current_user.is_authenticated:
return redirect(url_for('main.index'))
form = RegistrationForm()
if form.validate_on_submit():
user = User(username=form.username.data, email=form.email.data)
user.set_password(form.password.data)
db.session.add(user)
db.session.commit()
flash('Registration successful!', 'success')
return redirect(url_for('auth.login'))
return render_template('auth/register.html', form=form)
```
```python
# app/models.py
from app import db, login
from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import UserMixin
from datetime import datetime
class User(UserMixin, db.Model):
"""User model."""
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), index=True, unique=True)
email = db.Column(db.String(120), index=True, unique=True)
password_hash = db.Column(db.String(128))
created_at = db.Column(db.DateTime, index=True, default=datetime.utcnow)
posts = db.relationship('Post', backref='author', lazy='dynamic')
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def __repr__(self):
return f'<User {self.username}>'
@login.user_loader
def load_user(id):
return User.query.get(int(id))
class Post(db.Model):
"""Post model."""
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(100), nullable=False)
content = db.Column(db.Text, nullable=False)
published = db.Column(db.Boolean, default=False)
created_at = db.Column(db.DateTime, index=True, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
def __repr__(self):
return f'<Post {self.title}>'
```
```python
# app.py (main application entry point)
from app import create_app, db
from app.models import User, Post
app = create_app()
@app.shell_context_processor
def make_shell_context():
"""Add models to shell context."""
return {'db': db, 'User': User, 'Post': Post}
if __name__ == '__main__':
app.run(debug=True)
```
## Project Structure
```
project/
├── app/
│ ├── __init__.py # Application factory
│ ├── config.py # Configuration classes
│ ├── models.py # Database models
│ ├── main/
│ │ ├── __init__.py # Main blueprint
│ │ └── routes.py # Main routes
│ ├── auth/
│ │ ├── __init__.py # Auth blueprint
│ │ ├── routes.py # Auth routes
│ │ └── forms.py # Auth forms
│ ├── api/
│ │ ├── __init__.py # API blueprint
│ │ └── routes.py # API routes
│ └── templates/ # Jinja2 templates
├── migrations/ # Database migrations
├── logs/ # Application logs
├── app.py # Application entry point
├── requirements.txt # Dependencies
└── .env # Environment variables
```
## Benefits of Application Factory
- **Testing**: Easy to create app instances with different configs
- **Scalability**: Modular design with blueprints
- **Configuration**: Environment-specific settings
- **Extensions**: Proper initialization order
- **Maintainability**: Clear separation of concerns
@@ -0,0 +1,243 @@
# Flask Blueprint Generator
Create organized Flask blueprints for modular application structure.
## Usage
```bash
# Create a new blueprint
flask create-blueprint users
flask create-blueprint api/v1
```
## Blueprint Structure
Generates a complete blueprint with:
- Routes and view functions
- Error handlers
- Template folder structure
- Static file organization
## Example Blueprint
```python
# app/blueprints/users/__init__.py
from flask import Blueprint
users_bp = Blueprint(
'users',
__name__,
url_prefix='/users',
template_folder='templates',
static_folder='static'
)
from . import routes, models
# app/blueprints/users/routes.py
from flask import render_template, request, redirect, url_for, flash
from . import users_bp
from .models import User
from .forms import UserForm
@users_bp.route('/')
def index():
"""List all users."""
users = User.query.all()
return render_template('users/index.html', users=users)
@users_bp.route('/create', methods=['GET', 'POST'])
def create():
"""Create a new user."""
form = UserForm()
if form.validate_on_submit():
user = User(
username=form.username.data,
email=form.email.data
)
user.save()
flash('User created successfully!', 'success')
return redirect(url_for('users.index'))
return render_template('users/create.html', form=form)
@users_bp.route('/<int:user_id>')
def detail(user_id):
"""Show user details."""
user = User.query.get_or_404(user_id)
return render_template('users/detail.html', user=user)
@users_bp.route('/<int:user_id>/edit', methods=['GET', 'POST'])
def edit(user_id):
"""Edit an existing user."""
user = User.query.get_or_404(user_id)
form = UserForm(obj=user)
if form.validate_on_submit():
user.username = form.username.data
user.email = form.email.data
user.save()
flash('User updated successfully!', 'success')
return redirect(url_for('users.detail', user_id=user.id))
return render_template('users/edit.html', form=form, user=user)
@users_bp.route('/<int:user_id>/delete', methods=['POST'])
def delete(user_id):
"""Delete a user."""
user = User.query.get_or_404(user_id)
user.delete()
flash('User deleted successfully!', 'success')
return redirect(url_for('users.index'))
# Error handlers
@users_bp.errorhandler(404)
def not_found(error):
return render_template('users/404.html'), 404
@users_bp.errorhandler(500)
def internal_error(error):
return render_template('users/500.html'), 500
```
## Blueprint Models
```python
# app/blueprints/users/models.py
from app.extensions import db
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
class User(db.Model):
"""User model."""
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
email = db.Column(db.String(120), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
is_active = db.Column(db.Boolean, default=True)
def __repr__(self):
return f'<User {self.username}>'
def set_password(self, password):
"""Set password hash."""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""Check password hash."""
return check_password_hash(self.password_hash, password)
def save(self):
"""Save user to database."""
db.session.add(self)
db.session.commit()
def delete(self):
"""Delete user from database."""
db.session.delete(self)
db.session.commit()
def to_dict(self):
"""Convert to dictionary."""
return {
'id': self.id,
'username': self.username,
'email': self.email,
'created_at': self.created_at.isoformat(),
'is_active': self.is_active
}
```
## Blueprint Forms
```python
# app/blueprints/users/forms.py
from flask_wtf import FlaskForm
from wtforms import StringField, EmailField, PasswordField, BooleanField
from wtforms.validators import DataRequired, Email, Length, EqualTo
from .models import User
class UserForm(FlaskForm):
"""User creation/edit form."""
username = StringField(
'Username',
validators=[
DataRequired(),
Length(min=3, max=80)
]
)
email = EmailField(
'Email',
validators=[
DataRequired(),
Email(),
Length(max=120)
]
)
password = PasswordField(
'Password',
validators=[
DataRequired(),
Length(min=8)
]
)
confirm_password = PasswordField(
'Confirm Password',
validators=[
DataRequired(),
EqualTo('password', message='Passwords must match')
]
)
is_active = BooleanField('Active')
def validate_username(self, field):
"""Validate username uniqueness."""
if User.query.filter_by(username=field.data).first():
raise ValidationError('Username already exists.')
def validate_email(self, field):
"""Validate email uniqueness."""
if User.query.filter_by(email=field.data).first():
raise ValidationError('Email already registered.')
```
## Registration in Main App
```python
# app/__init__.py
from flask import Flask
from app.blueprints.users import users_bp
def create_app():
app = Flask(__name__)
# Register blueprints
app.register_blueprint(users_bp)
return app
```
## Template Structure
```
templates/
├── base.html
└── users/
├── index.html
├── create.html
├── detail.html
├── edit.html
├── 404.html
└── 500.html
```
## Best Practices
- Use blueprints to organize related functionality
- Keep models, forms, and routes in separate files
- Implement proper error handling
- Use URL prefixes for namespacing
- Follow RESTful routing conventions
- Include comprehensive docstrings
- Add form validation and CSRF protection
@@ -0,0 +1,410 @@
# Flask Database Management
Complete database setup and management for Flask applications using SQLAlchemy.
## Usage
```bash
# Initialize database
flask db init
# Create migration
flask db migrate -m "Initial migration"
# Apply migrations
flask db upgrade
# Downgrade migration
flask db downgrade
```
## Database Configuration
```python
# config.py
import os
from urllib.parse import quote_plus
class Config:
"""Base configuration."""
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_RECORD_QUERIES = True
class DevelopmentConfig(Config):
"""Development configuration."""
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
'sqlite:///app.db'
class ProductionConfig(Config):
"""Production configuration."""
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
f"postgresql://{os.environ.get('DB_USER')}:{quote_plus(os.environ.get('DB_PASSWORD'))}@" \
f"{os.environ.get('DB_HOST')}:{os.environ.get('DB_PORT', '5432')}/{os.environ.get('DB_NAME')}"
class TestingConfig(Config):
"""Testing configuration."""
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig,
'default': DevelopmentConfig
}
```
## Database Extensions
```python
# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_wtf.csrf import CSRFProtect
from flask_caching import Cache
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
# Initialize extensions
db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
csrf = CSRFProtect()
cache = Cache()
limiter = Limiter(key_func=get_remote_address)
def init_extensions(app):
"""Initialize Flask extensions."""
db.init_app(app)
migrate.init_app(app, db)
login_manager.init_app(app)
csrf.init_app(app)
cache.init_app(app)
limiter.init_app(app)
# Configure login manager
login_manager.login_view = 'auth.login'
login_manager.login_message = 'Please log in to access this page.'
login_manager.login_message_category = 'info'
```
## Base Model
```python
# app/models/base.py
from app.extensions import db
from datetime import datetime
from sqlalchemy.ext.declarative import declared_attr
class TimestampMixin:
"""Add timestamp fields to model."""
created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
class BaseModel(db.Model, TimestampMixin):
"""Base model with common functionality."""
__abstract__ = True
id = db.Column(db.Integer, primary_key=True)
@declared_attr
def __tablename__(cls):
return cls.__name__.lower()
def save(self, commit=True):
"""Save model to database."""
db.session.add(self)
if commit:
db.session.commit()
return self
def delete(self, commit=True):
"""Delete model from database."""
db.session.delete(self)
if commit:
db.session.commit()
def update(self, **kwargs):
"""Update model attributes."""
for key, value in kwargs.items():
if hasattr(self, key):
setattr(self, key, value)
return self.save()
def to_dict(self, exclude=None):
"""Convert model to dictionary."""
exclude = exclude or []
return {
column.name: getattr(self, column.name)
for column in self.__table__.columns
if column.name not in exclude
}
@classmethod
def get_or_404(cls, id):
"""Get model by ID or raise 404."""
return cls.query.get_or_404(id)
@classmethod
def create(cls, **kwargs):
"""Create new model instance."""
instance = cls(**kwargs)
return instance.save()
```
## Example Models
```python
# app/models/user.py
from app.extensions import db
from app.models.base import BaseModel
from werkzeug.security import generate_password_hash, check_password_hash
from flask_login import UserMixin
class User(UserMixin, BaseModel):
"""User model."""
__tablename__ = 'users'
username = db.Column(db.String(80), unique=True, nullable=False, index=True)
email = db.Column(db.String(120), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(255), nullable=False)
first_name = db.Column(db.String(50), nullable=False)
last_name = db.Column(db.String(50), nullable=False)
is_active = db.Column(db.Boolean, default=True, nullable=False)
is_admin = db.Column(db.Boolean, default=False, nullable=False)
last_login = db.Column(db.DateTime)
# Relationships
posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
def __repr__(self):
return f'<User {self.username}>'
def set_password(self, password):
"""Set password hash."""
self.password_hash = generate_password_hash(password)
def check_password(self, password):
"""Check password hash."""
return check_password_hash(self.password_hash, password)
@property
def full_name(self):
"""Get user's full name."""
return f"{self.first_name} {self.last_name}"
def to_dict(self, exclude=None):
"""Convert to dictionary excluding sensitive data."""
exclude = exclude or ['password_hash']
return super().to_dict(exclude=exclude)
# app/models/post.py
class Post(BaseModel):
"""Blog post model."""
__tablename__ = 'posts'
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text, nullable=False)
slug = db.Column(db.String(200), unique=True, nullable=False, index=True)
status = db.Column(db.String(20), default='draft', nullable=False)
published_at = db.Column(db.DateTime)
# Foreign keys
user_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
category_id = db.Column(db.Integer, db.ForeignKey('categories.id'))
# Relationships
category = db.relationship('Category', backref='posts')
tags = db.relationship('Tag', secondary='post_tags', backref='posts')
def __repr__(self):
return f'<Post {self.title}>'
@property
def is_published(self):
"""Check if post is published."""
return self.status == 'published' and self.published_at is not None
```
## Database CLI Commands
```python
# app/cli.py
import click
from flask import current_app
from flask.cli import with_appcontext
from app.extensions import db
from app.models import User, Post, Category
@click.command()
@with_appcontext
def init_db():
"""Initialize database."""
db.create_all()
click.echo('Database initialized.')
@click.command()
@with_appcontext
def seed_db():
"""Seed database with sample data."""
# Create admin user
admin = User(
username='admin',
email='admin@example.com',
first_name='Admin',
last_name='User',
is_admin=True
)
admin.set_password('admin123')
admin.save()
# Create sample category
category = Category(
name='Technology',
description='Tech-related posts'
)
category.save()
# Create sample post
post = Post(
title='Welcome to Flask',
content='This is a sample blog post.',
slug='welcome-to-flask',
status='published',
user_id=admin.id,
category_id=category.id
)
post.save()
click.echo('Database seeded with sample data.')
@click.command()
@with_appcontext
def reset_db():
"""Reset database."""
if click.confirm('Are you sure you want to reset the database?'):
db.drop_all()
db.create_all()
click.echo('Database reset.')
def init_commands(app):
"""Register CLI commands."""
app.cli.add_command(init_db)
app.cli.add_command(seed_db)
app.cli.add_command(reset_db)
```
## Connection Pooling
```python
# app/database.py
from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool
def configure_database(app):
"""Configure database with connection pooling."""
if app.config.get('SQLALCHEMY_DATABASE_URI', '').startswith('postgresql'):
# PostgreSQL configuration
engine = create_engine(
app.config['SQLALCHEMY_DATABASE_URI'],
poolclass=QueuePool,
pool_size=10,
max_overflow=20,
pool_recycle=3600,
pool_pre_ping=True
)
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {
'pool_size': 10,
'max_overflow': 20,
'pool_recycle': 3600,
'pool_pre_ping': True
}
```
## Database Utilities
```python
# app/utils/database.py
from app.extensions import db
from sqlalchemy import text
from flask import current_app
def execute_sql(sql, params=None):
"""Execute raw SQL query."""
with db.engine.connect() as conn:
result = conn.execute(text(sql), params or {})
return result.fetchall()
def backup_database():
"""Create database backup."""
# Implementation depends on database type
pass
def check_database_health():
"""Check database connection health."""
try:
db.session.execute(text('SELECT 1'))
return True
except Exception as e:
current_app.logger.error(f'Database health check failed: {e}')
return False
def get_table_info(table_name):
"""Get table information."""
inspector = db.inspect(db.engine)
return {
'columns': inspector.get_columns(table_name),
'indexes': inspector.get_indexes(table_name),
'foreign_keys': inspector.get_foreign_keys(table_name)
}
```
## Testing Database
```python
# tests/conftest.py
import pytest
from app import create_app
from app.extensions import db
from app.models import User
@pytest.fixture(scope='session')
def app():
"""Create test app."""
app = create_app('testing')
with app.app_context():
db.create_all()
yield app
db.drop_all()
@pytest.fixture
def client(app):
"""Create test client."""
return app.test_client()
@pytest.fixture
def db_session(app):
"""Create database session for testing."""
with app.app_context():
db.session.begin()
yield db.session
db.session.rollback()
@pytest.fixture
def user(db_session):
"""Create test user."""
user = User(
username='testuser',
email='test@example.com',
first_name='Test',
last_name='User'
)
user.set_password('testpass')
user.save()
return user
```
@@ -0,0 +1,620 @@
# Flask Deployment Configuration
Complete production deployment setup for Flask applications.
## Usage
```bash
# Build Docker image
docker build -t myapp .
# Run with Docker Compose
docker-compose up -d
# Deploy to cloud
gunicorn --bind 0.0.0.0:8000 app:app
```
## Production Configuration
```python
# config.py
import os
from urllib.parse import quote_plus
class ProductionConfig:
"""Production configuration."""
# Security
SECRET_KEY = os.environ.get('SECRET_KEY')
DEBUG = False
TESTING = False
# Database
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
f"postgresql://{os.environ.get('DB_USER')}:{quote_plus(os.environ.get('DB_PASSWORD'))}@" \
f"{os.environ.get('DB_HOST')}:{os.environ.get('DB_PORT', '5432')}/{os.environ.get('DB_NAME')}"
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ENGINE_OPTIONS = {
'pool_size': 10,
'max_overflow': 20,
'pool_recycle': 3600,
'pool_pre_ping': True
}
# Security Headers
SECURITY_HEADERS = {
'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
'X-Content-Type-Options': 'nosniff',
'X-Frame-Options': 'DENY',
'X-XSS-Protection': '1; mode=block',
'Content-Security-Policy': "default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline'"
}
# Session
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
PERMANENT_SESSION_LIFETIME = 3600 # 1 hour
# Cache
CACHE_TYPE = 'redis'
CACHE_REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
CACHE_DEFAULT_TIMEOUT = 300
# Rate Limiting
RATELIMIT_STORAGE_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/1')
RATELIMIT_DEFAULT = '100/hour'
# Logging
LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')
LOG_FILE = os.environ.get('LOG_FILE', '/var/log/app/app.log')
# File Upload
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB
UPLOAD_FOLDER = os.environ.get('UPLOAD_FOLDER', '/var/uploads')
# Email
MAIL_SERVER = os.environ.get('MAIL_SERVER')
MAIL_PORT = int(os.environ.get('MAIL_PORT', 587))
MAIL_USE_TLS = os.environ.get('MAIL_USE_TLS', 'true').lower() in ['true', 'on', '1']
MAIL_USERNAME = os.environ.get('MAIL_USERNAME')
MAIL_PASSWORD = os.environ.get('MAIL_PASSWORD')
MAIL_DEFAULT_SENDER = os.environ.get('MAIL_DEFAULT_SENDER')
```
## WSGI Configuration
```python
# wsgi.py
import os
from app import create_app
# Get environment
config_name = os.environ.get('FLASK_ENV', 'production')
app = create_app(config_name)
if __name__ == "__main__":
app.run()
```
## Gunicorn Configuration
```python
# gunicorn.conf.py
import multiprocessing
import os
# Server socket
bind = f"0.0.0.0:{os.environ.get('PORT', 8000)}"
backlog = 2048
# Worker processes
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = 'sync'
worker_connections = 1000
timeout = 30
keepalive = 60
max_requests = 1000
max_requests_jitter = 100
# Security
limit_request_line = 4094
limit_request_fields = 100
limit_request_field_size = 8190
# Logging
accesslog = '-'
errorlog = '-'
loglevel = os.environ.get('LOG_LEVEL', 'info').lower()
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'
# Process naming
proc_name = 'flask_app'
# Server mechanics
daemon = False
pidfile = '/tmp/gunicorn.pid'
user = os.environ.get('USER', 'www-data')
group = os.environ.get('GROUP', 'www-data')
tmp_upload_dir = None
# SSL
keyfile = os.environ.get('SSL_KEYFILE')
certfile = os.environ.get('SSL_CERTFILE')
```
## Docker Configuration
```dockerfile
# Dockerfile
FROM python:3.11-slim
# Set environment variables
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_NO_CACHE_DIR=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1
# Install system dependencies
RUN apt-get update && apt-get install -y \
build-essential \
libpq-dev \
curl \
&& rm -rf /var/lib/apt/lists/*
# Create app user
RUN groupadd -r appuser && useradd -r -g appuser appuser
# Set work directory
WORKDIR /app
# Install Python dependencies
COPY requirements/production.txt ./requirements.txt
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Create necessary directories
RUN mkdir -p /var/log/app /var/uploads && \
chown -R appuser:appuser /app /var/log/app /var/uploads
# Switch to non-root user
USER appuser
# Expose port
EXPOSE 8000
# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Run application
CMD ["gunicorn", "--config", "gunicorn.conf.py", "wsgi:app"]
```
## Docker Compose
```yaml
# docker-compose.yml
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
environment:
- FLASK_ENV=production
- DATABASE_URL=postgresql://postgres:password@db:5432/myapp
- REDIS_URL=redis://redis:6379/0
- SECRET_KEY=${SECRET_KEY}
depends_on:
- db
- redis
volumes:
- uploads:/var/uploads
- logs:/var/log/app
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
db:
image: postgres:15
environment:
- POSTGRES_DB=myapp
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=password
volumes:
- postgres_data:/var/lib/postgresql/data
restart: unless-stopped
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 30s
timeout: 10s
retries: 3
redis:
image: redis:7-alpine
command: redis-server --appendonly yes
volumes:
- redis_data:/data
restart: unless-stopped
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 30s
timeout: 10s
retries: 3
nginx:
image: nginx:alpine
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx.conf:/etc/nginx/nginx.conf:ro
- ./ssl:/etc/nginx/ssl:ro
- uploads:/var/uploads:ro
depends_on:
- web
restart: unless-stopped
volumes:
postgres_data:
redis_data:
uploads:
logs:
```
## Nginx Configuration
```nginx
# nginx.conf
events {
worker_connections 1024;
}
http {
upstream app {
server web:8000;
}
# Rate limiting
limit_req_zone $binary_remote_addr zone=api:10m rate=10r/s;
limit_req_zone $binary_remote_addr zone=login:10m rate=1r/s;
# SSL configuration
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers ECDHE-RSA-AES128-GCM-SHA256:ECDHE-RSA-AES256-GCM-SHA384;
ssl_prefer_server_ciphers off;
# Security headers
add_header X-Frame-Options DENY;
add_header X-Content-Type-Options nosniff;
add_header X-XSS-Protection "1; mode=block";
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains" always;
server {
listen 80;
server_name example.com www.example.com;
return 301 https://$server_name$request_uri;
}
server {
listen 443 ssl http2;
server_name example.com www.example.com;
ssl_certificate /etc/nginx/ssl/cert.pem;
ssl_certificate_key /etc/nginx/ssl/key.pem;
# File upload size
client_max_body_size 16M;
# Static files
location /static/ {
alias /var/uploads/static/;
expires 1y;
add_header Cache-Control "public, immutable";
}
location /uploads/ {
alias /var/uploads/;
expires 1h;
}
# API rate limiting
location /api/ {
limit_req zone=api burst=20 nodelay;
proxy_pass http://app;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# Login rate limiting
location /auth/login {
limit_req zone=login burst=5 nodelay;
proxy_pass http://app;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# Main application
location / {
proxy_pass http://app;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Timeout settings
proxy_connect_timeout 60s;
proxy_send_timeout 60s;
proxy_read_timeout 60s;
}
}
}
```
## Environment Variables
```bash
# .env.production
# Application
FLASK_ENV=production
SECRET_KEY=your-super-secret-key-here
# Database
DATABASE_URL=postgresql://user:password@localhost:5432/myapp
DB_HOST=localhost
DB_PORT=5432
DB_NAME=myapp
DB_USER=user
DB_PASSWORD=password
# Redis
REDIS_URL=redis://localhost:6379/0
# Email
MAIL_SERVER=smtp.gmail.com
MAIL_PORT=587
MAIL_USE_TLS=true
MAIL_USERNAME=your-email@gmail.com
MAIL_PASSWORD=your-app-password
MAIL_DEFAULT_SENDER=your-email@gmail.com
# Logging
LOG_LEVEL=INFO
LOG_FILE=/var/log/app/app.log
# File Upload
UPLOAD_FOLDER=/var/uploads
# SSL (if using)
SSL_KEYFILE=/path/to/private.key
SSL_CERTFILE=/path/to/certificate.crt
```
## Health Check Endpoint
```python
# app/health.py
from flask import Blueprint, jsonify
from app.extensions import db
from sqlalchemy import text
import redis
import os
health_bp = Blueprint('health', __name__)
@health_bp.route('/health')
def health_check():
"""Application health check."""
checks = {
'status': 'healthy',
'database': check_database(),
'redis': check_redis(),
'disk_space': check_disk_space()
}
# Determine overall status
if all(check['status'] == 'ok' for check in checks.values() if isinstance(check, dict)):
status_code = 200
else:
status_code = 503
checks['status'] = 'unhealthy'
return jsonify(checks), status_code
def check_database():
"""Check database connectivity."""
try:
db.session.execute(text('SELECT 1'))
return {'status': 'ok', 'message': 'Database connection successful'}
except Exception as e:
return {'status': 'error', 'message': str(e)}
def check_redis():
"""Check Redis connectivity."""
try:
redis_url = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
r = redis.from_url(redis_url)
r.ping()
return {'status': 'ok', 'message': 'Redis connection successful'}
except Exception as e:
return {'status': 'error', 'message': str(e)}
def check_disk_space():
"""Check available disk space."""
try:
import shutil
total, used, free = shutil.disk_usage('/')
free_percent = (free / total) * 100
if free_percent > 10:
status = 'ok'
elif free_percent > 5:
status = 'warning'
else:
status = 'critical'
return {
'status': status,
'free_space_percent': round(free_percent, 2),
'free_space_gb': round(free / (1024**3), 2)
}
except Exception as e:
return {'status': 'error', 'message': str(e)}
```
## Monitoring and Logging
```python
# app/logging.py
import logging
import logging.handlers
import os
from flask import request, g
import time
def setup_logging(app):
"""Setup application logging."""
if not app.debug and not app.testing:
# File logging
if app.config.get('LOG_FILE'):
file_handler = logging.handlers.RotatingFileHandler(
app.config['LOG_FILE'],
maxBytes=10240000, # 10MB
backupCount=10
)
file_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s '
'[in %(pathname)s:%(lineno)d]'
))
file_handler.setLevel(getattr(logging, app.config.get('LOG_LEVEL', 'INFO')))
app.logger.addHandler(file_handler)
# Console logging
if not app.logger.handlers:
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s'
))
stream_handler.setLevel(logging.INFO)
app.logger.addHandler(stream_handler)
app.logger.setLevel(logging.INFO)
app.logger.info('Application startup')
# Request timing middleware
@app.before_request
def before_request():
g.start_time = time.time()
@app.after_request
def after_request(response):
if hasattr(g, 'start_time'):
duration = time.time() - g.start_time
app.logger.info(
f'{request.method} {request.path} - '
f'{response.status_code} - {duration:.3f}s'
)
return response
```
## Database Backup Script
```bash
#!/bin/bash
# backup.sh
set -e
# Configuration
BACKUP_DIR="/var/backups/db"
DATABASE_URL="$DATABASE_URL"
DATE=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="$BACKUP_DIR/backup_$DATE.sql"
RETENTION_DAYS=7
# Create backup directory
mkdir -p "$BACKUP_DIR"
# Create backup
echo "Creating database backup..."
pg_dump "$DATABASE_URL" > "$BACKUP_FILE"
# Compress backup
gzip "$BACKUP_FILE"
# Remove old backups
echo "Cleaning up old backups..."
find "$BACKUP_DIR" -name "backup_*.sql.gz" -mtime +$RETENTION_DAYS -delete
echo "Backup completed: $BACKUP_FILE.gz"
```
## Deployment Scripts
```bash
#!/bin/bash
# deploy.sh
set -e
echo "Starting deployment..."
# Pull latest code
git pull origin main
# Build new Docker image
docker-compose build web
# Run database migrations
docker-compose run --rm web flask db upgrade
# Update services
docker-compose up -d
# Health check
echo "Waiting for application to be ready..."
sleep 10
if curl -f http://localhost:8000/health; then
echo "Deployment successful!"
else
echo "Deployment failed - health check failed"
exit 1
fi
```
## Monitoring with Prometheus
```python
# app/metrics.py
from prometheus_flask_exporter import PrometheusMetrics
from flask import request
import time
def setup_metrics(app):
"""Setup Prometheus metrics."""
metrics = PrometheusMetrics(app)
# Custom metrics
metrics.info('app_info', 'Application info', version='1.0.0')
# Database connection pool metrics
@metrics.gauge('db_pool_size', 'Database connection pool size')
def db_pool_size():
from app.extensions import db
return db.engine.pool.size()
return metrics
```
@@ -0,0 +1,217 @@
# Flask Route Generator
Create Flask routes with proper structure and error handling.
## Purpose
This command helps you quickly create Flask routes with validation, error handling, and best practices.
## Usage
```
/flask-route
```
## What this command does
1. **Creates route functions** with proper decorators
2. **Adds request validation** and error handling
3. **Includes JSON responses** and status codes
4. **Implements authentication** if needed
5. **Follows Flask conventions** and best practices
## Example Output
```python
# routes.py or app.py
from flask import Flask, request, jsonify, abort
from flask_sqlalchemy import SQLAlchemy
from werkzeug.exceptions import BadRequest
app = Flask(__name__)
@app.route('/users', methods=['GET'])
def get_users():
"""Get all users with optional pagination."""
try:
page = request.args.get('page', 1, type=int)
per_page = request.args.get('per_page', 10, type=int)
users = User.query.paginate(
page=page,
per_page=per_page,
error_out=False
)
return jsonify({
'users': [user.to_dict() for user in users.items],
'total': users.total,
'pages': users.pages,
'current_page': page
}), 200
except Exception as e:
return jsonify({'error': 'Failed to fetch users'}), 500
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
"""Get a specific user by ID."""
try:
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict()), 200
except Exception as e:
return jsonify({'error': 'User not found'}), 404
@app.route('/users', methods=['POST'])
def create_user():
"""Create a new user."""
try:
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
# Validate required fields
required_fields = ['name', 'email']
for field in required_fields:
if field not in data:
return jsonify({'error': f'{field} is required'}), 400
# Check if email already exists
if User.query.filter_by(email=data['email']).first():
return jsonify({'error': 'Email already exists'}), 409
# Create new user
user = User(
name=data['name'],
email=data['email'],
phone=data.get('phone'),
address=data.get('address')
)
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
except BadRequest:
return jsonify({'error': 'Invalid JSON data'}), 400
except Exception as e:
db.session.rollback()
return jsonify({'error': 'Failed to create user'}), 500
@app.route('/users/<int:user_id>', methods=['PUT'])
def update_user(user_id):
"""Update an existing user."""
try:
user = User.query.get_or_404(user_id)
data = request.get_json()
if not data:
return jsonify({'error': 'No data provided'}), 400
# Update fields
if 'name' in data:
user.name = data['name']
if 'email' in data:
# Check if new email already exists
existing_user = User.query.filter_by(email=data['email']).first()
if existing_user and existing_user.id != user_id:
return jsonify({'error': 'Email already exists'}), 409
user.email = data['email']
if 'phone' in data:
user.phone = data['phone']
if 'address' in data:
user.address = data['address']
db.session.commit()
return jsonify(user.to_dict()), 200
except BadRequest:
return jsonify({'error': 'Invalid JSON data'}), 400
except Exception as e:
db.session.rollback()
return jsonify({'error': 'Failed to update user'}), 500
@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
"""Delete a user."""
try:
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return jsonify({'message': 'User deleted successfully'}), 200
except Exception as e:
db.session.rollback()
return jsonify({'error': 'Failed to delete user'}), 500
# Error handlers
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': 'Resource not found'}), 404
@app.errorhandler(400)
def bad_request(error):
return jsonify({'error': 'Bad request'}), 400
@app.errorhandler(500)
def internal_error(error):
return jsonify({'error': 'Internal server error'}), 500
```
## Route Patterns Supported
### Basic Routes
```python
@app.route('/')
@app.route('/users')
@app.route('/users/<int:user_id>')
```
### HTTP Methods
```python
@app.route('/users', methods=['GET', 'POST'])
@app.route('/users/<int:id>', methods=['GET', 'PUT', 'DELETE'])
```
### URL Parameters
```python
@app.route('/users/<int:user_id>')
@app.route('/posts/<string:slug>')
@app.route('/files/<path:filename>')
```
## Best Practices Included
- **Input validation** for all user data
- **Proper HTTP status codes** (200, 201, 400, 404, 500)
- **JSON responses** with consistent structure
- **Error handling** with try/catch blocks
- **Database rollback** on errors
- **RESTful conventions** for URL design
- **Documentation strings** for each route
- **Request data validation** before processing
## Common Response Patterns
```python
# Success with data
return jsonify({'data': result}), 200
# Created resource
return jsonify({'data': new_resource, 'id': new_id}), 201
# Validation error
return jsonify({'error': 'Field is required'}), 400
# Not found
return jsonify({'error': 'Resource not found'}), 404
# Server error
return jsonify({'error': 'Internal server error'}), 500
```
@@ -0,0 +1,559 @@
# Flask Testing Suite
Comprehensive testing setup for Flask applications with pytest.
## Usage
```bash
# Run all tests
pytest
# Run with coverage
pytest --cov=app --cov-report=html
# Run specific test file
pytest tests/test_models.py
# Run with verbose output
pytest -v
```
## Test Configuration
```python
# pytest.ini
[tool:pytest]
testpaths = tests
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts =
--cov=app
--cov-report=term-missing
--cov-report=html:htmlcov
--strict-markers
--disable-warnings
markers =
unit: Unit tests
integration: Integration tests
slow: Slow running tests
auth: Authentication tests
```
## Test Fixtures
```python
# tests/conftest.py
import pytest
import tempfile
import os
from app import create_app
from app.extensions import db
from app.models import User, Post, Category
from flask_login import login_user
@pytest.fixture(scope='session')
def app():
"""Create test application."""
# Create temporary database
db_fd, db_path = tempfile.mkstemp()
app = create_app({
'TESTING': True,
'SQLALCHEMY_DATABASE_URI': f'sqlite:///{db_path}',
'WTF_CSRF_ENABLED': False,
'SECRET_KEY': 'test-secret-key'
})
with app.app_context():
db.create_all()
yield app
# Cleanup
os.close(db_fd)
os.unlink(db_path)
@pytest.fixture
def client(app):
"""Create test client."""
return app.test_client()
@pytest.fixture
def runner(app):
"""Create test CLI runner."""
return app.test_cli_runner()
@pytest.fixture
def db_session(app):
"""Create database session for testing."""
with app.app_context():
connection = db.engine.connect()
transaction = connection.begin()
# Configure session to use the connection
db.session.configure(bind=connection)
yield db.session
# Rollback transaction
transaction.rollback()
connection.close()
db.session.remove()
@pytest.fixture
def user(db_session):
"""Create test user."""
user = User(
username='testuser',
email='test@example.com',
first_name='Test',
last_name='User'
)
user.set_password('testpass123')
user.save()
return user
@pytest.fixture
def admin_user(db_session):
"""Create admin user."""
admin = User(
username='admin',
email='admin@example.com',
first_name='Admin',
last_name='User',
is_admin=True
)
admin.set_password('adminpass123')
admin.save()
return admin
@pytest.fixture
def category(db_session):
"""Create test category."""
category = Category(
name='Test Category',
description='A test category'
)
category.save()
return category
@pytest.fixture
def post(db_session, user, category):
"""Create test post."""
post = Post(
title='Test Post',
content='This is a test post content.',
slug='test-post',
status='published',
user_id=user.id,
category_id=category.id
)
post.save()
return post
@pytest.fixture
def auth_headers(user):
"""Create authentication headers."""
# For API testing
token = user.generate_auth_token()
return {'Authorization': f'Bearer {token}'}
```
## Model Testing
```python
# tests/test_models.py
import pytest
from datetime import datetime
from app.models import User, Post, Category
from werkzeug.security import check_password_hash
class TestUser:
"""Test User model."""
def test_user_creation(self, db_session):
"""Test user creation."""
user = User(
username='newuser',
email='new@example.com',
first_name='New',
last_name='User'
)
user.set_password('password123')
user.save()
assert user.id is not None
assert user.username == 'newuser'
assert user.email == 'new@example.com'
assert user.full_name == 'New User'
assert user.is_active is True
assert user.is_admin is False
assert user.created_at is not None
def test_password_hashing(self, user):
"""Test password hashing."""
user.set_password('newpassword')
assert user.password_hash != 'newpassword'
assert check_password_hash(user.password_hash, 'newpassword')
assert user.check_password('newpassword')
assert not user.check_password('wrongpassword')
def test_user_repr(self, user):
"""Test user string representation."""
assert repr(user) == '<User testuser>'
def test_user_to_dict(self, user):
"""Test user dictionary conversion."""
user_dict = user.to_dict()
assert 'username' in user_dict
assert 'email' in user_dict
assert 'password_hash' not in user_dict # Should be excluded
def test_user_relationships(self, user, post):
"""Test user relationships."""
assert post in user.posts
assert user.posts.count() == 1
class TestPost:
"""Test Post model."""
def test_post_creation(self, db_session, user, category):
"""Test post creation."""
post = Post(
title='New Post',
content='New post content',
slug='new-post',
status='draft',
user_id=user.id,
category_id=category.id
)
post.save()
assert post.id is not None
assert post.title == 'New Post'
assert post.author == user
assert post.category == category
assert not post.is_published
def test_published_status(self, post):
"""Test post published status."""
assert post.is_published # Published with published_at
post.status = 'draft'
assert not post.is_published
```
## View Testing
```python
# tests/test_views.py
import pytest
from flask import url_for
from app.models import User
class TestMainViews:
"""Test main application views."""
def test_home_page(self, client):
"""Test home page."""
response = client.get('/')
assert response.status_code == 200
assert b'Welcome' in response.data
def test_about_page(self, client):
"""Test about page."""
response = client.get('/about')
assert response.status_code == 200
class TestUserViews:
"""Test user-related views."""
def test_user_list(self, client, user):
"""Test user list page."""
response = client.get('/users/')
assert response.status_code == 200
assert user.username.encode() in response.data
def test_user_detail(self, client, user):
"""Test user detail page."""
response = client.get(f'/users/{user.id}')
assert response.status_code == 200
assert user.username.encode() in response.data
def test_user_create_get(self, client):
"""Test user creation form."""
response = client.get('/users/create')
assert response.status_code == 200
assert b'Create User' in response.data
def test_user_create_post(self, client, db_session):
"""Test user creation submission."""
data = {
'username': 'newuser',
'email': 'new@example.com',
'first_name': 'New',
'last_name': 'User',
'password': 'password123',
'confirm_password': 'password123'
}
response = client.post('/users/create', data=data, follow_redirects=True)
assert response.status_code == 200
# Check user was created
user = User.query.filter_by(username='newuser').first()
assert user is not None
assert user.email == 'new@example.com'
def test_user_edit(self, client, user):
"""Test user editing."""
data = {
'username': user.username,
'email': 'updated@example.com',
'first_name': 'Updated',
'last_name': 'User'
}
response = client.post(f'/users/{user.id}/edit', data=data, follow_redirects=True)
assert response.status_code == 200
# Refresh user from database
db_session.refresh(user)
assert user.email == 'updated@example.com'
assert user.first_name == 'Updated'
```
## Authentication Testing
```python
# tests/test_auth.py
import pytest
from flask import url_for
from app.models import User
class TestAuthentication:
"""Test authentication functionality."""
def test_login_page(self, client):
"""Test login page access."""
response = client.get('/auth/login')
assert response.status_code == 200
assert b'Login' in response.data
def test_valid_login(self, client, user):
"""Test valid user login."""
data = {
'username': user.username,
'password': 'testpass123'
}
response = client.post('/auth/login', data=data, follow_redirects=True)
assert response.status_code == 200
assert b'Welcome' in response.data
def test_invalid_login(self, client, user):
"""Test invalid login credentials."""
data = {
'username': user.username,
'password': 'wrongpassword'
}
response = client.post('/auth/login', data=data)
assert response.status_code == 200
assert b'Invalid' in response.data
def test_logout(self, client, user):
"""Test user logout."""
# Login first
with client.session_transaction() as sess:
sess['_user_id'] = str(user.id)
response = client.get('/auth/logout', follow_redirects=True)
assert response.status_code == 200
def test_register_page(self, client):
"""Test registration page."""
response = client.get('/auth/register')
assert response.status_code == 200
assert b'Register' in response.data
def test_valid_registration(self, client, db_session):
"""Test valid user registration."""
data = {
'username': 'newuser',
'email': 'new@example.com',
'first_name': 'New',
'last_name': 'User',
'password': 'password123',
'confirm_password': 'password123'
}
response = client.post('/auth/register', data=data, follow_redirects=True)
assert response.status_code == 200
# Check user was created
user = User.query.filter_by(username='newuser').first()
assert user is not None
```
## API Testing
```python
# tests/test_api.py
import pytest
import json
from flask import url_for
class TestUserAPI:
"""Test User API endpoints."""
def test_get_users(self, client, user):
"""Test GET /api/users."""
response = client.get('/api/users')
assert response.status_code == 200
data = json.loads(response.data)
assert 'users' in data
assert len(data['users']) >= 1
def test_get_user(self, client, user):
"""Test GET /api/users/<id>."""
response = client.get(f'/api/users/{user.id}')
assert response.status_code == 200
data = json.loads(response.data)
assert data['username'] == user.username
assert data['email'] == user.email
def test_create_user(self, client, db_session):
"""Test POST /api/users."""
user_data = {
'username': 'apiuser',
'email': 'api@example.com',
'first_name': 'API',
'last_name': 'User',
'password': 'password123'
}
response = client.post(
'/api/users',
data=json.dumps(user_data),
content_type='application/json'
)
assert response.status_code == 201
data = json.loads(response.data)
assert data['username'] == 'apiuser'
def test_update_user(self, client, user, auth_headers):
"""Test PUT /api/users/<id>."""
update_data = {
'email': 'updated@example.com',
'first_name': 'Updated'
}
response = client.put(
f'/api/users/{user.id}',
data=json.dumps(update_data),
content_type='application/json',
headers=auth_headers
)
assert response.status_code == 200
data = json.loads(response.data)
assert data['email'] == 'updated@example.com'
def test_delete_user(self, client, user, auth_headers):
"""Test DELETE /api/users/<id>."""
response = client.delete(
f'/api/users/{user.id}',
headers=auth_headers
)
assert response.status_code == 204
```
## Test Utilities
```python
# tests/utils.py
import json
from flask import url_for
def login_user(client, username, password):
"""Helper to login user in tests."""
return client.post('/auth/login', data={
'username': username,
'password': password
}, follow_redirects=True)
def logout_user(client):
"""Helper to logout user in tests."""
return client.get('/auth/logout', follow_redirects=True)
def assert_json_response(response, expected_status=200):
"""Assert JSON response format and status."""
assert response.status_code == expected_status
assert response.content_type == 'application/json'
return json.loads(response.data)
def create_test_data(db_session):
"""Create common test data."""
from app.models import User, Category, Post
# Create test users
users = []
for i in range(3):
user = User(
username=f'user{i}',
email=f'user{i}@example.com',
first_name=f'User{i}',
last_name='Test'
)
user.set_password('password123')
user.save()
users.append(user)
return {'users': users}
```
## Performance Testing
```python
# tests/test_performance.py
import pytest
import time
from app.models import User
@pytest.mark.slow
class TestPerformance:
"""Test application performance."""
def test_user_query_performance(self, db_session):
"""Test user query performance."""
# Create multiple users
users = []
for i in range(100):
user = User(
username=f'perfuser{i}',
email=f'perf{i}@example.com',
first_name=f'Perf{i}',
last_name='User'
)
users.append(user)
db_session.bulk_save_objects(users)
db_session.commit()
# Test query performance
start_time = time.time()
result = User.query.all()
end_time = time.time()
assert len(result) >= 100
assert (end_time - start_time) < 0.1 # Should complete in under 100ms
def test_endpoint_response_time(self, client):
"""Test endpoint response time."""
start_time = time.time()
response = client.get('/')
end_time = time.time()
assert response.status_code == 200
assert (end_time - start_time) < 0.5 # Should respond in under 500ms
```
@@ -0,0 +1,391 @@
# Flask Project Configuration
This file provides specific guidance for Flask web application development using Claude Code.
## Project Overview
This is a Flask web application project optimized for scalable web development with the Flask micro-framework. The project follows Flask best practices and modern Python development patterns.
## Flask-Specific Development Commands
### Project Management
- `flask run` - Start development server
- `flask run --host=0.0.0.0 --port=5000` - Start server accessible from network
- `flask shell` - Open Flask shell with application context
- `python -m flask --help` - Show available Flask commands
### Database Management
- `flask db init` - Initialize database migrations
- `flask db migrate -m "message"` - Create database migration
- `flask db upgrade` - Apply database migrations
- `flask db downgrade` - Rollback database migration
- `flask db current` - Show current migration
- `flask db history` - Show migration history
### Development Tools
- `flask routes` - Show all registered routes
- `flask --version` - Show Flask version
- `export FLASK_ENV=development` - Set development environment
- `export FLASK_DEBUG=1` - Enable debug mode
### Custom Commands
- `flask init-db` - Initialize database with tables
- `flask seed-db` - Seed database with sample data
- `flask reset-db` - Reset database (development only)
## Flask Project Structure
```
myproject/
├── app/ # Application package
│ ├── __init__.py # Application factory
│ ├── extensions.py # Flask extensions
│ ├── config.py # Configuration settings
│ ├── models/ # Database models
│ │ ├── __init__.py
│ │ ├── base.py # Base model class
│ │ ├── user.py # User model
│ │ └── post.py # Post model
│ ├── blueprints/ # Application blueprints
│ │ ├── __init__.py
│ │ ├── main/ # Main blueprint
│ │ ├── auth/ # Authentication blueprint
│ │ ├── api/ # API blueprint
│ │ └── admin/ # Admin blueprint
│ ├── templates/ # Jinja2 templates
│ │ ├── base.html
│ │ ├── index.html
│ │ └── auth/
│ ├── static/ # Static files
│ │ ├── css/
│ │ ├── js/
│ │ └── images/
│ ├── forms/ # WTForms
│ │ ├── __init__.py
│ │ ├── auth.py
│ │ └── user.py
│ ├── utils/ # Utility functions
│ └── cli.py # Custom CLI commands
├── migrations/ # Database migrations
├── tests/ # Test files
│ ├── conftest.py
│ ├── test_models.py
│ ├── test_views.py
│ └── test_api.py
├── requirements/ # Requirements files
│ ├── base.txt
│ ├── development.txt
│ └── production.txt
├── wsgi.py # WSGI entry point
├── gunicorn.conf.py # Gunicorn configuration
└── docker-compose.yml # Docker Compose configuration
```
## Flask Application Factory
```python
# app/__init__.py
from flask import Flask
from app.extensions import db, migrate, login_manager, csrf, cache
from app.config import config
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
login_manager.init_app(app)
csrf.init_app(app)
cache.init_app(app)
# Register blueprints
from app.blueprints.main import main_bp
from app.blueprints.auth import auth_bp
from app.blueprints.api import api_bp
app.register_blueprint(main_bp)
app.register_blueprint(auth_bp, url_prefix='/auth')
app.register_blueprint(api_bp, url_prefix='/api/v1')
# Register CLI commands
from app.cli import init_commands
init_commands(app)
return app
```
## Configuration Management
```python
# app/config.py
import os
from datetime import timedelta
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY') or 'dev-secret-key'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_RECORD_QUERIES = True
# Session configuration
PERMANENT_SESSION_LIFETIME = timedelta(hours=1)
SESSION_COOKIE_SECURE = True
SESSION_COOKIE_HTTPONLY = True
SESSION_COOKIE_SAMESITE = 'Lax'
# File upload
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB
UPLOAD_FOLDER = os.path.join(os.getcwd(), 'uploads')
# Cache
CACHE_TYPE = 'simple'
CACHE_DEFAULT_TIMEOUT = 300
class DevelopmentConfig(Config):
DEBUG = True
SQLALCHEMY_DATABASE_URI = os.environ.get('DEV_DATABASE_URL') or \
'sqlite:///dev.db'
SESSION_COOKIE_SECURE = False
class ProductionConfig(Config):
DEBUG = False
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL')
# Security headers
SECURITY_HEADERS = {
'Strict-Transport-Security': 'max-age=31536000; includeSubDomains',
'X-Content-Type-Options': 'nosniff',
'X-Frame-Options': 'DENY',
'X-XSS-Protection': '1; mode=block'
}
class TestingConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
WTF_CSRF_ENABLED = False
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig,
'default': DevelopmentConfig
}
```
## Flask Best Practices
### Application Structure
- Use application factory pattern for configuration flexibility
- Organize code into blueprints for modularity
- Separate models, views, and forms into different modules
- Use extensions.py to initialize Flask extensions
- Implement proper error handling and logging
### Database Models
- Use SQLAlchemy ORM for database operations
- Implement base model with common functionality
- Add proper relationships between models
- Use database migrations for schema changes
- Implement model validation and constraints
### Blueprint Organization
- Group related functionality into blueprints
- Use URL prefixes for namespacing
- Implement blueprint-specific templates
- Add proper error handlers for each blueprint
- Use blueprint factories for complex blueprints
### Template Management
- Use template inheritance for consistent layout
- Create reusable template macros
- Implement proper CSRF protection in forms
- Use Flask-WTF for form handling and validation
- Organize templates by blueprint
### Security Considerations
- Always validate and sanitize user input
- Use Flask-Login for user session management
- Implement proper authentication and authorization
- Use CSRF protection for all forms
- Set secure session cookie configuration
- Implement rate limiting for API endpoints
## Flask Extensions
### Essential Extensions
```python
# app/extensions.py
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager
from flask_wtf.csrf import CSRFProtect
from flask_caching import Cache
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
db = SQLAlchemy()
migrate = Migrate()
login_manager = LoginManager()
csrf = CSRFProtect()
cache = Cache()
limiter = Limiter(key_func=get_remote_address)
```
### Recommended Extensions
- **Flask-SQLAlchemy** - Database ORM
- **Flask-Migrate** - Database migrations
- **Flask-Login** - User session management
- **Flask-WTF** - Form handling and CSRF protection
- **Flask-Caching** - Caching support
- **Flask-Limiter** - Rate limiting
- **Flask-Mail** - Email support
- **Flask-Admin** - Admin interface
## Testing Strategy
### Test Organization
```python
# tests/conftest.py
import pytest
from app import create_app
from app.extensions import db
@pytest.fixture(scope='session')
def app():
app = create_app('testing')
with app.app_context():
db.create_all()
yield app
db.drop_all()
@pytest.fixture
def client(app):
return app.test_client()
@pytest.fixture
def runner(app):
return app.test_cli_runner()
```
### Test Types
- **Unit tests** for models and utilities
- **Integration tests** for views and API endpoints
- **Functional tests** for user workflows
- **Performance tests** for critical paths
### Testing Best Practices
- Use fixtures for common test data
- Test both success and error conditions
- Mock external dependencies
- Use factory_boy for test data generation
- Implement database transaction rollback in tests
## Performance Optimization
### Database Optimization
- Use connection pooling for production
- Implement query optimization with indexes
- Use lazy loading for relationships
- Cache frequently accessed data
- Monitor database query performance
### Caching Strategy
- Implement Redis for session storage
- Use view-level caching for static content
- Cache database query results
- Implement cache invalidation strategies
- Use CDN for static files
### Application Optimization
- Use Gunicorn with multiple workers
- Implement proper logging and monitoring
- Optimize static file serving
- Use async tasks for long-running operations
- Implement proper error handling
## Deployment Considerations
### Production Setup
- Use environment variables for configuration
- Implement proper logging and monitoring
- Set up database connection pooling
- Configure reverse proxy (Nginx)
- Use HTTPS with proper SSL certificates
### Docker Configuration
```dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements/production.txt requirements.txt
RUN pip install -r requirements.txt
COPY . .
EXPOSE 5000
CMD ["gunicorn", "--config", "gunicorn.conf.py", "wsgi:app"]
```
### Environment Variables
```bash
FLASK_ENV=production
SECRET_KEY=your-secret-key
DATABASE_URL=postgresql://user:pass@host:port/db
REDIS_URL=redis://host:port/db
```
## Common Flask Patterns
### Custom Decorators
```python
from functools import wraps
from flask import abort
from flask_login import current_user
def admin_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if not current_user.is_admin:
abort(403)
return f(*args, **kwargs)
return decorated_function
```
### Request Context Processors
```python
@app.context_processor
def inject_user():
return dict(current_user=current_user)
```
### Custom Filters
```python
@app.template_filter('datetime')
def datetime_filter(value, format='%Y-%m-%d %H:%M'):
return value.strftime(format) if value else ''
```
## Development Workflow
### Getting Started
1. Clone the repository
2. Create virtual environment: `python -m venv venv`
3. Activate environment: `source venv/bin/activate`
4. Install dependencies: `pip install -r requirements/development.txt`
5. Set environment variables
6. Initialize database: `flask db upgrade`
7. Run development server: `flask run`
### Development Process
1. Create feature branch from main
2. Implement changes with tests
3. Run test suite: `pytest`
4. Check code quality: `flake8`, `black`
5. Create pull request for review
6. Deploy after approval
### Code Quality Tools
- **Black** - Code formatting
- **isort** - Import sorting
- **flake8** - Linting
- **mypy** - Type checking
- **pytest** - Testing framework
- **coverage** - Test coverage