Install Alembic with pip:
pip install alembic
Setup Process
- Initialize repository:
alembic init migrations - Create ORM models (example):
from sqlalchemy import Column, Integer, String, create_engine, Text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class Customer(Base):
__tablename__ = 'customers'
id = Column(Integer, primary_key=True)
name = Column(String(50), nullable=False)
email = Column(String(100), unique=True)
- Configure database connection in
alembic.ini:
sqlalchemy.url = postgresql+psycopg2://user:password@localhost/mydb
- Update
env.pyto include model metadata:
import os
import sys
sys.path.append(os.path.dirname(os.path.abspath(__file__)) + "/../")
from models import Base
target_metadata = Base.metadata
- Generate migration files:
alembic revision --autogenerate -m "Initial schema" - Apply migrations:
alembic upgrade head
Create uwsgi.ini with deployment settings:
[uwsgi]
socket = 127.0.0.1:9000
pythonpath = /opt/myproject
module = app
callable = create_app
processes = 4
threads = 2
daemonize = /var/log/uwsgi.log
Start the server:
uwsgi --ini uwsgi.ini
Nginx configuration example:
location /api {
include uwsgi_params;
uwsgi_pass 127.0.0.1:9000;
}
Index Creation
__table_args__ = (
Index("idx_product_category", 'category_id'),
Index("idx_product_price", 'price')
)
Efficient Queries
from sqlalchemy import and_
from sqlalchemy.orm import joinedload
# Eager loading to prevent N+1 queries
results = session.query(Order).options(joinedload(Order.items)).all()
# Batch operations
session.bulk_insert_mappings(Customer, customer_data)
Database Locking
from sqlalchemy import select
# Row-level locking
stmt = select(User).with_for_update(nowait=True)
user = session.execute(stmt).scalar_one()
Basic Application Structure
from flask import Flask, render_template
app = Flask(__name__)
@app.route('/')
def home():
return render_template('index.html', title='Home Page')
Blueprint Organization
from flask import Blueprint
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
@admin_bp.route('/dashboard')
def dashboard():
return "Admin Dashboard"
Session Management
from flask import session
@app.route('/login', methods=['POST'])
def login():
session['user_id'] = user.id
return redirect('/dashboard')
Model Definition
from django.db import models
class Product(models.Model):
name = models.CharField(max_length=100)
price = models.DecimalField(max_digits=10, decimal_places=2)
created_at = models.DateTimeField(auto_now_add=True)
View Implementation
from django.shortcuts import render
from .models import Product
def product_list(request):
products = Product.objects.all()
return render(request, 'products/list.html', {'products': products})
Middleware Example
class AuthenticationMiddleware:
def __init__(self, get_response):
self.get_response = get_response
def __call__(self, request):
if not request.user.is_authenticated and request.path != '/login/':
return redirect('/login/')
return self.get_response(request)
Basic Server Settup
import tornado.ioloop
import tornado.web
class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, Tornado!")
def make_app():
return tornado.web.Application([
(r"/", MainHandler),
])
if __name__ == "__main__":
app = make_app()
app.listen(8888)
tornado.ioloop.IOLoop.current().start()
Asynchronous Request Handling
import tornado.gen
class AsyncHandler(tornado.web.RequestHandler):
@tornado.gen.coroutine
def get(self):
http_client = tornado.httpclient.AsyncHTTPClient()
response = yield http_client.fetch("http://example.com")
self.write(response.body)
Application Setup
from blacksheep import Application
import uvicorn
app = Application()
@app.route("/items/{item_id}")
async def get_item(item_id: int):
return {"item_id": item_id}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=5555)
WebSocket Implementation
from blacksheep import WebSocket
@app.websocket("/ws/{client_id}")
async def websocket_handler(websocket: WebSocket, client_id: str):
await websocket.accept()
await websocket.send_text(f"Welcome, client {client_id}!")
while True:
message = await websocket.receive_text()
await websocket.send_text(f"Received: {message}")