Master Flask‑SQLAlchemy: Connect, Model, and Query MySQL Seamlessly

Introduction

When developing web applications we often need to interact with a database. This article introduces Flask‑SQLAlchemy, a plugin that simplifies database operations in Flask.

1. Installation and Import

pip install flask-sqlalchemy
from flask_sqlalchemy import SQLAlchemy
import os
import pymysql as p
from flask import Flask

2. Basic Usage

2.1 Connect to MySQL

2.1.1 Write configuration in a separate file

db_type='mysql'
db_conn='pymysql'
host='127.0.0.1'
username='root'
password='123456'
port='3306'
db_name='people'
SQLALCHEMY_DATABASE_URI='{}+{}://{}:{}@{}:{}/{}?charset=utf8'.format(db_type,db_conn,username,password,host,port,db_name)
SQLALCHEMY_COMMIT_ON_TEARDOWN=False
SQLALCHEMY_TRACK_MODIFICATIONS=True

Load the configuration in the Flask application:

from flask_sqlalchemy import SQLAlchemy
from flask import Flask
import config
app = Flask(__name__)
app.config.from_object(config)  # load config file
db = SQLAlchemy(app)
db.create_all()  # create all tables

2.1.2 Directly write configuration in the app

app = Flask(__name__)
p.install_as_MySQLdb()  # required for MySQLdb on Python 3
app.config['SECRET_KEY'] = os.urandom(50)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:[email protected]:3306/people'
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True  # auto‑commit after each request
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True
db = SQLAlchemy(app)

2.1.3 Additional configuration parameters

SQLALCHEMY_NATIVE_UNICODE    # disable native unicode support
SQLALCHEMY_POOL_SIZE         # size of the connection pool (default 5)
SQLALCHEMY_POOL_TIMEOUT      # pool timeout in seconds (default 10)
SQLALCHEMY_POOL_RECYCLE      # recycle connections after given seconds (MySQL default 2 h)
SQLALCHEMY_RECORD_QUERIES    # enable/disable query recording
SQLALCHEMY_ECHO              # echo SQL statements for debugging
SQLALCHEMY_BINDS             # dictionary mapping bind keys to database URIs

2.1.4 Create engine manually

from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://root:[email protected]:3306/people', echo=True)
engine.connect()

2.1.5 Connect to multiple databases

app.config['SQLALCHEMY_BINDS'] = {
    'users': 'mysql+pymysql://root:[email protected]/user',
    'admin': 'sqlite:///C:/Users/Administrator/Desktop/admin',
    'buy':   'postgresql://root:[email protected]/buy'
}
# create tables for a specific bind
db.create_all(bind=['users'])
db.create_all(bind='admin')
__bind_key__ = 'users'  # set bind key inside model's __table_args__ if needed

3. Define Models and Create Tables

3.1 Inherit from db.Model

from flask_sqlalchemy import SQLAlchemy
from flask import Flask
import config
app = Flask(__name__)
app.config.from_object(config)
db = SQLAlchemy(app)
class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(100), nullable=False)
    password = db.Column(db.String(100), nullable=False)
    def __init__(self, username, password):
        self.username = username
        self.password = password
    def __repr__(self):
        return '%s is %r' % (self.__class__.__name__, self.username)
@app.before_first_request
def create_table():
    db.drop_all()
    db.create_all()
@app.route('/')
def create():
    u1 = User('hwhrr', '123321')
    u2 = User('hwzjj', '123456')
    u3 = User('hwzzz', '654321')
    db.session.add_all([u1, u2, u3])
    db.session.commit()
    return u1.__repr__()
if __name__ == '__main__':
    app.run(debug=True)

3.2 Relationships

One‑to‑One

class Father(db.Model):
    __tablename__ = 'Father'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    son_fa = db.relationship('Son', uselist=False, back_populates='fa_son')
    def __repr__(self):
        return '%s is %r' % (self.__class__.__name__, self.name)
class Son(db.Model):
    __tablename__ = 'Son'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    fa_son = db.relationship('Father', back_populates='son_fa')
    def __repr__(self):
        return '%s is %r' % (self.__class__.__name__, self.name)

One‑to‑Many

class Father(db.Model):
    __tablename__ = 'Father'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    son_fa = db.relationship('Son', backref='fa', lazy='dynamic')
class Son(db.Model):
    __tablename__ = 'Son'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    f_id = db.Column(db.Integer, db.ForeignKey('Father.id'))
    def __repr__(self):
        return '%s is %r' % (self.__class__.__name__, self.name)

Many‑to‑One

class Father(db.Model):
    __tablename__ = 'Father'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    son_fa = db.relationship('Son', back_populates='fa_son')
class Son(db.Model):
    __tablename__ = 'Son'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    f_id = db.Column(db.Integer, db.ForeignKey('Father.id'))
    fa_son = db.relationship('Father', back_populates='son_fa')
    def __repr__(self):
        return '%s is %r' % (self.__class__.__name__, self.name)

Many‑to‑Many

# association table
gl = db.Table('glb',
    db.Column('id', db.Integer, primary_key=True, autoincrement=True),
    db.Column('son_id', db.Integer, db.ForeignKey('Son.id')),
    db.Column('father_id', db.Integer, db.ForeignKey('Father.id'))
)
class Father(db.Model):
    __tablename__ = 'Father'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    son_fa = db.relationship('Son', secondary=gl, backref='fas', lazy='dynamic')
    def __repr__(self):
        return '%s is %r' % (self.__class__.__name__, self.name)
class Son(db.Model):
    __tablename__ = 'Son'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(100), nullable=False)
    age = db.Column(db.Integer, nullable=False)
    def __repr__(self):
        return '%s is %r' % (self.__class__.__name__, self.name)

4. Query Data

4.1 Retrieve all records

Father.query.all()

4.2 Filter by exact value

Father.query.filter_by(name='hw').all()

4.3 Fuzzy filter

Father.query.filter(Father.name.startswith('h')).all()

4.4 Primary‑key lookup

Father.query.get(1)

4.5 Negation

from sqlalchemy import not_
Father.query.filter(not_(Father.name == 'hw')).all()

4.6 Conjunction

from sqlalchemy import and_
Father.query.filter(and_(Father.name.endswith('h'), Father.age == 50)).all()

4.7 Disjunction

from sqlalchemy import or_
Father.query.filter(or_(Father.name.endswith('h'), Father.age == 50)).all()

4.8 One‑to‑many forward query

Son.query.filter_by(f_id=2).all()

4.9 One‑to‑many reverse query

Son.query.filter_by(fa=use1).all()

4.10 First matching record

Son.query.filter(Son.age == 10).first()
Son.query.filter(Son.age == 10)[0]

4.11 Limit results

Son.query.filter(Son.age == 10).limit(10).all()

4.12 Offset results

Son.query.filter(Son.age == 10).offset(2).all()

4.13 Order by descending

Son.query.filter(Son.age == 10).order_by(Son.create_time.desc()).all()

4.14 Group by

Son.query.filter(Son.age == 10).group_by(Son.name).all()

4.15 Count

Son.query.filter(Son.age == 10).count()

4.16 Slice

Son.query.slice(10, 40).all()
Son.query[10:40]

4.17 Pagination

p = request.args.get('p')
paginate = Son.query.paginate(p=int(p), per_page=3)
# paginate.pages, paginate.page, paginate.has_next, paginate.items, etc.

4.18 Retrieve a single record

Son.query.one()

5. Update Data

ss = Son.query.get(1)
ss.name = 'fasd'
db.session.commit()

6. Delete Data

ss = Son.query.get(1)
db.session.delete(ss)
db.session.commit()

Conclusion

SQLAlchemy supports a wide range of table definitions and operations. By mastering its usage you can easily manipulate database data from a Flask backend, enabling interactive front‑end pages and rapid development of small web sites.