Building a Web Paint App with Flask and MongoDB: A Complete Tutorial
Building web applications that handle dynamic, schema-less data often requires moving beyond traditional relational …
Building upon our foundational Flask MongoDB paint application, this comprehensive guide explores modern Flask development practices for creating scalable, production-ready web applications. We’ll cover advanced patterns, security implementations, and deployment strategies essential for professional web development.
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)
This modern Flask development guide provides production-ready patterns for building scalable web applications. The architecture shown supports authentication, caching, rate limiting, and comprehensive API documentation essential for professional web services.
For more advanced web development concepts, explore our related tutorials on MongoDB integration patterns and real-time web applications.