Flask Multiple Databases

In more complex Flask applications, you might need to work with multiple databases simultaneously. This could be for various reasons, such as:

• Separating user data from application data
• Integrating with legacy systems that use different database engines
• Implementing a microservice architecture with specialized databases
• Creating read replicas for performance improvements

This tutorial will show you how to configure and use multiple databases in your Flask application using Flask-SQLAlchemy.

Prerequisites

Before diving in, make sure you're familiar with:

• Basic Flask application structure
• Flask-SQLAlchemy fundamentals
• Database connections and ORM concepts

Setting Up Multiple Databases in Flask

Step 1: Install Required Packages

First, ensure you have the necessary packages installed:

bash

pip install flask flask-sqlalchemy psycopg2-binary pymysql

Step 2: Configure Multiple Database URIs

In your Flask application, you'll need to configure multiple database URIs:

python

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# Configure the primary database (PostgreSQL)
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://username:password@localhost/primary_db'

# Configure additional databases
app.config['SQLALCHEMY_BINDS'] = {
    'users_db': 'postgresql://username:password@localhost/users_db',
    'logs_db': 'mysql://username:password@localhost/logs_db'
}

# Disable track modifications warning
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# Initialize SQLAlchemy
db = SQLAlchemy(app)

The SQLALCHEMY_DATABASE_URI sets your primary database, while SQLALCHEMY_BINDS is a dictionary that maps names to additional database URIs.

Step 3: Define Models for Different Databases

Next, define your models and associate them with specific databases:

python

# Model for the primary database (default)
class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)
    
    def __repr__(self):
        return f'<Product {self.name}>'

# Model for the users_db
class User(db.Model):
    __bind_key__ = 'users_db'  # This associates the model with users_db
    
    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)
    
    def __repr__(self):
        return f'<User {self.username}>'

# Model for the logs_db
class Log(db.Model):
    __bind_key__ = 'logs_db'  # This associates the model with logs_db
    
    id = db.Column(db.Integer, primary_key=True)
    timestamp = db.Column(db.DateTime, nullable=False, default=db.func.current_timestamp())
    level = db.Column(db.String(10), nullable=False)
    message = db.Column(db.Text, nullable=False)
    
    def __repr__(self):
        return f'<Log {self.timestamp} {self.level}>'

The __bind_key__ class attribute is what connects your model to a specific database from the SQLALCHEMY_BINDS configuration.

Step 4: Create the Tables

You'll need to create tables in each of your databases:

python

with app.app_context():
    # Create tables in the primary database
    db.create_all()
    
    # Create tables in the bound databases
    db.create_all(bind='users_db')
    db.create_all(bind='logs_db')

Working with Multiple Databases

Inserting Data

Here's how you can insert data into your different databases:

python

with app.app_context():
    # Add a product to the primary database
    new_product = Product(name='Laptop', price=999.99)
    db.session.add(new_product)
    
    # Add a user to the users_db
    new_user = User(username='john_doe', email='[email protected]')
    db.session.add(new_user)
    
    # Add a log to the logs_db
    new_log = Log(level='INFO', message='Application started')
    db.session.add(new_log)
    
    # Commit all changes
    db.session.commit()

Querying Data

Querying works the same way as with a single database:

python

with app.app_context():
    # Query from the primary database
    products = Product.query.all()
    print("Products:", products)
    
    # Query from the users_db
    users = User.query.all()
    print("Users:", users)
    
    # Query from the logs_db
    logs = Log.query.filter_by(level='INFO').all()
    print("Info Logs:", logs)

Using Sessions with Specific Binds

If you need more control, you can create sessions bound to specific databases:

python

with app.app_context():
    # Create a session for the users_db
    users_session = db.create_scoped_session(options={"bind": db.get_engine(app, bind="users_db")})
    
    # Use the specific session
    new_user = User(username='jane_doe', email='[email protected]')
    users_session.add(new_user)
    users_session.commit()
    users_session.close()

Real-World Example: E-commerce Application

Let's build a more complete example of an e-commerce application that uses multiple databases:

python

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

app = Flask(__name__)

# Database configuration
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgresql://username:password@localhost/ecommerce_products'
app.config['SQLALCHEMY_BINDS'] = {
    'users': 'postgresql://username:password@localhost/ecommerce_users',
    'orders': 'postgresql://username:password@localhost/ecommerce_orders',
    'analytics': 'mysql://username:password@localhost/ecommerce_analytics'
}
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

# Models for different databases
class Product(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(100), nullable=False)
    price = db.Column(db.Float, nullable=False)
    stock = db.Column(db.Integer, nullable=False)
    category = db.Column(db.String(50), nullable=False)

class User(db.Model):
    __bind_key__ = '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)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

class Order(db.Model):
    __bind_key__ = 'orders'
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, nullable=False)
    total_amount = db.Column(db.Float, nullable=False)
    status = db.Column(db.String(20), nullable=False, default='pending')
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

class OrderItem(db.Model):
    __bind_key__ = 'orders'
    id = db.Column(db.Integer, primary_key=True)
    order_id = db.Column(db.Integer, nullable=False)
    product_id = db.Column(db.Integer, nullable=False)
    quantity = db.Column(db.Integer, nullable=False)
    price = db.Column(db.Float, nullable=False)

class PageView(db.Model):
    __bind_key__ = 'analytics'
    id = db.Column(db.Integer, primary_key=True)
    page = db.Column(db.String(200), nullable=False)
    user_id = db.Column(db.Integer, nullable=True)
    timestamp = db.Column(db.DateTime, default=datetime.utcnow)
    ip_address = db.Column(db.String(50), nullable=True)

# Create tables
with app.app_context():
    db.create_all()
    db.create_all(bind='users')
    db.create_all(bind='orders')
    db.create_all(bind='analytics')

# API Routes
@app.route('/products', methods=['GET'])
def get_products():
    products = Product.query.all()
    result = [
        {'id': product.id, 'name': product.name, 
         'price': product.price, 'stock': product.stock} 
        for product in products
    ]
    
    # Log this page view to analytics database
    page_view = PageView(page='/products', ip_address=request.remote_addr)
    db.session.add(page_view)
    db.session.commit()
    
    return jsonify(result)

@app.route('/users/<int:user_id>/orders', methods=['GET'])
def get_user_orders(user_id):
    # Check if user exists in users database
    user = User.query.get_or_404(user_id)
    
    # Get orders from orders database
    orders = Order.query.filter_by(user_id=user_id).all()
    result = [
        {'id': order.id, 'total': order.total_amount, 
         'status': order.status, 'date': order.created_at} 
        for order in orders
    ]
    
    # Log this page view to analytics database
    page_view = PageView(page=f'/users/{user_id}/orders', 
                         user_id=user_id,
                         ip_address=request.remote_addr)
    db.session.add(page_view)
    db.session.commit()
    
    return jsonify(result)

if __name__ == '__main__':
    app.run(debug=True)

This example demonstrates:

• A product catalog database (primary)
• A separate users database for customer information
• An orders database for transactional data
• An analytics database for tracking user behavior

Best Practices for Multiple Databases

1. Clearly define database responsibilities: Each database should have a specific purpose
2. Use consistent naming conventions: For both database keys and model names
3. Handle transactions carefully: Be aware that transactions are per database
4. Consider connection pooling: Manage your connection pools efficiently for multiple databases
5. Monitor database performance: Each database may have different performance characteristics

Common Challenges and Solutions

Challenge 1: Cross-Database Relationships

SQLAlchemy doesn't support relationships across different databases. To handle this, you'll need to manage these relationships in your application code:

python

def get_order_with_details(order_id):
    # Get the order from the orders database
    order = Order.query.get(order_id)
    if not order:
        return None
    
    # Get the user from the users database
    user = User.query.get(order.user_id)
    
    # Get order items and associated products
    order_items = OrderItem.query.filter_by(order_id=order_id).all()
    
    # Build the complete order data
    order_data = {
        'id': order.id,
        'user': {
            'id': user.id,
            'username': user.username
        },
        'total': order.total_amount,
        'status': order.status,
        'items': []
    }
    
    # Add each item with product details
    for item in order_items:
        product = Product.query.get(item.product_id)
        order_data['items'].append({
            'product_name': product.name,
            'quantity': item.quantity,
            'price': item.price
        })
    
    return order_data

Challenge 2: Managing Transactions Across Databases

When you need to make changes to multiple databases that must succeed or fail together:

python

def create_order(user_id, product_ids, quantities):
    # Start with getting products and calculating total
    products = []
    total_amount = 0
    
    for i, product_id in enumerate(product_ids):
        product = Product.query.get(product_id)
        if not product or product.stock < quantities[i]:
            return False, "Product unavailable"
        
        products.append(product)
        total_amount += product.price * quantities[i]
    
    try:
        # Create order in orders database
        order = Order(user_id=user_id, total_amount=total_amount)
        db.session.add(order)
        db.session.flush()  # This assigns an ID to order without committing
        
        # Add order items
        for i, product in enumerate(products):
            item = OrderItem(
                order_id=order.id,
                product_id=product.id,
                quantity=quantities[i],
                price=product.price
            )
            db.session.add(item)
        
        # Update stock in products database
        for i, product in enumerate(products):
            product.stock -= quantities[i]
        
        # Commit all changes
        db.session.commit()
        return True, order.id
        
    except Exception as e:
        db.session.rollback()
        return False, str(e)

Note that this approach isn't a true distributed transaction but uses SQLAlchemy's session to manage the operations.

Summary

Working with multiple databases in Flask offers flexibility for complex applications, allowing you to separate concerns and optimize for different use cases. With Flask-SQLAlchemy's binds feature, you can connect to different database engines and manage your models appropriately.

Key takeaways:

• Use SQLALCHEMY_BINDS to configure multiple database connections
• Assign models to specific databases with the __bind_key__ attribute
• Create tables in each database using db.create_all(bind='database_name')
• Handle transactions carefully when working across databases
• Consider application design patterns that work well with data partitioning

Additional Resources and Exercises

Resources

• Flask-SQLAlchemy Documentation on Binds
• SQLAlchemy Engine Configuration
• Database Sharding Patterns

Exercises

1. Basic Multiple Database Setup: Create a Flask application with two databases - one for products and another for reviews. Add functionality to display products with their associated reviews.

2. Read Replica Pattern: Implement a pattern where write operations go to a primary database, but read operations use a replica database. (You can simulate this with two database configurations.)

3. Database Migration Challenge: Create a Flask application that gradually migrates data from an old database schema to a new one, using both databases during the transition.

4. Analytics Database: Add an analytics database to an existing Flask application that tracks page views, user sessions, and conversion rates without affecting the performance of the main application.

5. Multi-tenant Application: Build a Flask application that serves multiple clients, each with their own separate database, while sharing common application code.