Reusable Python Code Modules, Part 5 - Database Client
Seamless Data Access: Building Reusable Database Client Modules
Database client modules are fundamental in building robust back-end APIs with Python and Flask. This post covers creating reusable database client modules, focusing on stability and reusability across projects.
Common Libraries and Tools
1. SQLAlchemy
Description: SQL toolkit and ORM for Python
Features: High-level ORM, low-level SQL execution, schema migrations with Alembic
Example:
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
db = SQLAlchemy(app)
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
@app.route('/create_user/<username>')
def create_user(username):
user = User(username=username)
db.session.add(user)
db.session.commit()
return 'User created'
2. Peewee
Description: Small, expressive ORM
Features: Simplicity, lightweight, supports SQLite, MySQL, PostgreSQL
Example:
from flask import Flask
from peewee import SqliteDatabase, Model, CharField
app = Flask(__name__)
db = SqliteDatabase('example.db')
class BaseModel(Model):
class Meta:
database = db
class User(BaseModel):
username = CharField(unique=True)
db.connect()
db.create_tables([User])
@app.route('/create_user/<username>')
def create_user(username):
user = User.create(username=username)
return 'User created'3. Pony ORM
Description: ORM with generator-based queries
Features: Entity-relationship diagrams, generator expressions for queries
Example:
from flask import Flask
from pony.orm import Database, Required, db_session
app = Flask(__name__)
db = Database()
db.bind(provider='sqlite', filename='example.db', create_db=True)
class User(db.Entity):
username = Required(str, unique=True)
db.generate_mapping(create_tables=True)
@app.route('/create_user/<username>')
@db_session
def create_user(username):
User(username=username)
return 'User created'4. Tortoise ORM
Description: Easy-to-use asyncio ORM inspired by Django
Features: Async support, migrations, Pydantic integration
Example:
from flask import Flask
from tortoise import Tortoise, fields
from tortoise.models import Model
import asyncio
app = Flask(__name__)
class User(Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=20, unique=True)
async def init():
await Tortoise.init(
db_url='sqlite://example.db',
modules={'models': ['__main__']}
)
await Tortoise.generate_schemas()
@app.route('/create_user/<username>')
async def create_user(username):
await User.create(username=username)
return 'User created'
if __name__ == '__main__':
asyncio.run(init())
app.run()Comparison of Libraries
SQLAlchemy vs. Peewee
SQLAlchemy: Highly flexible, supports complex queries, strong community support, can be overkill for small projects
Peewee: Simple, lightweight, easy to learn, limited for complex queries and large projects
Pony ORM vs. Tortoise ORM
Pony ORM: Intuitive queries with Python generators, good for small to medium projects, limited async support
Tortoise ORM: Full async support, integrates well with modern Python async frameworks, slightly steeper learning curve
Practical Examples
Database Client with SQLAlchemy
Description: Setting up and using SQLAlchemy in a Flask application
Example:
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
db = SQLAlchemy(app)
class User(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
db.create_all()
@app.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
new_user = User(username=data['username'])
db.session.add(new_user)
db.session.commit()
return jsonify({'message': 'User created'}), 201
@app.route('/users', methods=['GET'])
def get_users():
users = User.query.all()
return jsonify([user.username for user in users])
if __name__ == '__main__':
app.run()Database Client with Peewee
Description: Setting up and using Peewee in a Flask application
Example:
from flask import Flask, request, jsonify
from peewee import SqliteDatabase, Model, CharField
app = Flask(__name__)
db = SqliteDatabase('example.db')
class BaseModel(Model):
class Meta:
database = db
class User(BaseModel):
username = CharField(unique=True)
db.connect()
db.create_tables([User])
@app.route('/users', methods=['POST'])
def create_user():
data = request.get_json()
user = User.create(username=data['username'])
return jsonify({'message': 'User created'}), 201
@app.route('/users', methods=['GET'])
def get_users():
users = User.select()
return jsonify([user.username for user in users])
if __name__ == '__main__':
app.run()Database Client with Pony ORM
Description: Setting up and using Pony ORM in a Flask application
Example:
from flask import Flask, request, jsonify
from pony.orm import Database, Required, db_session
app = Flask(__name__)
db = Database()
db.bind(provider='sqlite', filename='example.db', create_db=True)
class User(db.Entity):
username = Required(str, unique=True)
db.generate_mapping(create_tables=True)
@app.route('/users', methods=['POST'])
@db_session
def create_user():
data = request.get_json()
User(username=data['username'])
return jsonify({'message': 'User created'}), 201
@app.route('/users', methods=['GET'])
@db_session
def get_users():
users = User.select()
return jsonify([user.username for user in users])
if __name__ == '__main__':
app.run()Database Client with Tortoise ORM
Description: Setting up and using Tortoise ORM in a Flask application
Example:
from flask import Flask, request, jsonify
from tortoise import Tortoise, fields
from tortoise.models import Model
import asyncio
app = Flask(__name__)
class User(Model):
id = fields.IntField(pk=True)
username = fields.CharField(max_length=20, unique=True)
async def init():
await Tortoise.init(
db_url='sqlite://example.db',
modules={'models': ['__main__']}
)
await Tortoise.generate_schemas()
@app.route('/users', methods=['POST'])
async def create_user():
data = request.get_json()
await User.create(username=data['username'])
return jsonify({'message': 'User created'}), 201
@app.route('/users', methods=['GET'])
async def get_users():
users = await User.all()
return jsonify([user.username for user in users])
if __name__ == '__main__':
asyncio.run(init())
app.run()