Flask
Flask Essentials
Flask Essentials App Setup & Routing from flask import Flask, request, jsonify, abort, g from flask_sqlalchemy import SQLAlchemy from functools import wraps imp…
Flask Essentials
App Setup & Routing
from flask import Flask, request, jsonify, abort, g
from flask_sqlalchemy import SQLAlchemy
from functools import wraps
import os
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('DATABASE_URL', 'sqlite:///dev.db')
app.config['SECRET_KEY'] = os.environ['SECRET_KEY']
db = SQLAlchemy(app)
# Routes
@app.route('/')
def index():
return jsonify({'message': 'Hello World'})
@app.route('/users', methods=['GET'])
def list_users():
page = request.args.get('page', 1, type=int)
limit = request.args.get('limit', 20, type=int)
users = User.query.paginate(page=page, per_page=limit, error_out=False)
return jsonify({
'data': [u.to_dict() for u in users.items],
'total': users.total,
'pages': users.pages,
})
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
user = User.query.get_or_404(user_id)
return jsonify(user.to_dict())
@app.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
if not data or not data.get('email'):
abort(400, description='Email is required')
user = User(email=data['email'], name=data.get('name', ''))
db.session.add(user)
db.session.commit()
return jsonify(user.to_dict()), 201
@app.route('/users/<int:user_id>', methods=['PATCH'])
def update_user(user_id):
user = User.query.get_or_404(user_id)
data = request.get_json()
if 'name' in data:
user.name = data['name']
db.session.commit()
return jsonify(user.to_dict())
@app.route('/users/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):
user = User.query.get_or_404(user_id)
db.session.delete(user)
db.session.commit()
return '', 204
# Error handlers
@app.errorhandler(404)
def not_found(error):
return jsonify({'error': 'Not found', 'message': str(error)}), 404
@app.errorhandler(400)
def bad_request(error):
return jsonify({'error': 'Bad request', 'message': str(error)}), 400Models & Blueprints
from datetime import datetime
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(255), unique=True, nullable=False)
name = db.Column(db.String(100), nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
is_active = db.Column(db.Boolean, default=True, nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
posts = db.relationship('Post', back_populates='author', lazy='dynamic')
def set_password(self, password):
from werkzeug.security import generate_password_hash
self.password_hash = generate_password_hash(password)
def check_password(self, password):
from werkzeug.security import check_password_hash
return check_password_hash(self.password_hash, password)
def to_dict(self):
return {'id': self.id, 'email': self.email, 'name': self.name,
'isActive': self.is_active, 'createdAt': self.created_at.isoformat()}
# Blueprints — organize routes into modules
from flask import Blueprint
users_bp = Blueprint('users', __name__, url_prefix='/api/users')
@users_bp.route('/')
def list_users(): ...
# Register in factory function
def create_app(config=None):
app = Flask(__name__)
app.config.from_object(config or 'config.Config')
db.init_app(app)
from .routes.users import users_bp
from .routes.posts import posts_bp
app.register_blueprint(users_bp)
app.register_blueprint(posts_bp)
return appMiddleware & Auth
import jwt
from functools import wraps
SECRET = app.config['SECRET_KEY']
def require_auth(f):
@wraps(f)
def decorated(*args, **kwargs):
auth = request.headers.get('Authorization', '')
if not auth.startswith('Bearer '):
abort(401, description='Missing token')
token = auth.split(' ')[1]
try:
payload = jwt.decode(token, SECRET, algorithms=['HS256'])
g.user_id = payload['sub']
except jwt.ExpiredSignatureError:
abort(401, description='Token expired')
except jwt.InvalidTokenError:
abort(401, description='Invalid token')
return f(*args, **kwargs)
return decorated
@app.route('/profile')
@require_auth
def profile():
user = User.query.get_or_404(g.user_id)
return jsonify(user.to_dict())
# Before/after request hooks
@app.before_request
def log_request():
g.start_time = time.time()
@app.after_request
def add_headers(response):
response.headers['X-Process-Time'] = str(time.time() - g.start_time)
return response
# Run
if __name__ == '__main__':
app.run(debug=True, port=5000)
# Production: gunicorn app:app --workers 4 --bind 0.0.0.0:8000