Flask Database Operations

In Flask, database operations are an important aspect of building web applications.

Flask provides multiple ways to interact with databases, including directly using SQL and utilizing ORM (Object-Relational Mapping) tools such as SQLAlchemy.

The following is a detailed explanation of Flask database operations, including basic operations using SQLAlchemy and direct execution of SQL statements.

  1. Using SQLAlchemy: Define models, configure the database, and perform basic CRUD operations.
  2. Creating and Managing Databases: Usedb.create_all()Create a table.
  3. CRUD Operations: Add, read, update, and delete records.
  4. Query Operations: Execute basic and complex queries, including sorting and pagination.
  5. Flask-Migrate: Use Flask-Migrate to manage database migrations.
  6. Executing Raw SQL: Use raw SQL statements for database operations.

1. Using SQLAlchemy

SQLAlchemy is a powerful ORM library that simplifies database operations by interacting with database tables through Python objects.

Flask-SQLAlchemy is a Flask extension for integrating SQLAlchemy.

Install Flask-SQLAlchemy

pip install flask-sqlalchemy

Configure SQLAlchemy

app.py file code:

Example

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'  # Use SQLite database
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

2. Define the model

A model is a Python class for a database table; each model class represents a table in the database.

Example

class User(db.Model):
    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}>'

db.Model: All model classes need to inherit from db.Model.

db.Column: Define the model's fields, specifying attributes such as field type, whether it is a primary key, whether it is unique, whether it can be null, etc.

3. Creating and Managing Databases

Create the database and tables

After defining the models, you can use the methods provided by SQLAlchemy to create the database and tables.

with app.app_context():
    db.create_all()

db.create_all(): Create tables corresponding to all models defined in the current context.

4. Basic CRUD Operations

Create Record

Example

@app.route('/add_user')
def add_user():
    new_user = User(username='john_doe', email='john@example.com')
    db.session.add(new_user)
    db.session.commit()
    return 'User added!'

db.session.add(new_user): Add the new user object to the session.

db.session.commit(): Commit the transaction to save changes to the database.

Read Record

Example

@app.route('/get_users')
def get_users():
    users = User.query.all()  # Get all users
    return '<br>'.join([f'{user.username} ({user.email})' for user in users])

User.query.all(): Query all user records.

Update Record

Example

@app.route('/update_user/<int:user_id>')
def update_user(user_id):
    user = User.query.get(user_id)
    if user:
        user.username = 'new_username'
        db.session.commit()
        return 'User updated!'
    return 'User not found!'

User.query.get(user_id): Query a single user record by primary key.

Update field values and commit the transaction.

Delete Record

Example

@app.route('/delete_user/<int:user_id>')
def delete_user(user_id):
    user = User.query.get(user_id)
    if user:
        db.session.delete(user)
        db.session.commit()
        return 'User deleted!'
    return 'User not found!'

db.session.delete(user): Delete the user record and commit the transaction.

5. Query operations

SQLAlchemy provides rich query functionality, allowing you to perform various query operations through query objects.

Basic Query

users = User.query.filter_by(username='john_doe').all()

filter_by(): Filter records by field value.

Complex Query

from sqlalchemy import or_

users = User.query.filter(or_(User.username == 'john_doe', User.email == 'john@example.com')).all()

or_(): Used for executing complex query conditions.

Sorting and Pagination

users = User.query.order_by(User.username).paginate(page=1, per_page=10)

order_by(): Sort by the specified field.

paginate(): Paginated query.

6. Use Flask-Migrate for migrations.

Flask-Migrate is an extension for database migrations, based on Alembic, which helps you manage database version control.

Install Flask-Migrate

pip install flask-migrate

Configure Flask-Migrate

app.py file code:

from flask_migrate import Migrate

migrate = Migrate(app, db)

Initialize Migration

flask db init

Create migration scripts

flask db migrate -m "Initial migration."

Apply Migration

flask db upgrade
  • flask db init: Initialize the migration environment.
  • flask db migrate -m "message": Create a migration script.
  • flask db upgrade: Apply migrations to the database.

7. Executing Raw SQL

Although SQLAlchemy provides ORM functionality, you can also execute raw SQL statements.

Example

@app.route('/raw_sql')
def raw_sql():
    result = db.session.execute('SELECT * FROM user')
    return '<br>'.join([str(row) for row in result])

db.session.execute(): Execute a raw SQL query.

Connecting to and operating a MySQL database in Flask

Connecting to and operating a MySQL database in Flask typically involves using SQLAlchemy or directly using Python drivers for MySQL. The following are detailed steps, including using Flask-SQLAlchemy and directly using Python drivers for MySQL.

1. Use Flask-SQLAlchemy to connect to MySQL

Flask-SQLAlchemy is a Flask extension that simplifies the configuration and operation of SQLAlchemy. To connect to MySQL, you need to install Flask-SQLAlchemy and a MySQL driver.

Install the required libraries:

pip install flask-sqlalchemy mysqlclient
  • flask-sqlalchemy: Flask's SQLAlchemy extension.
  • mysqlclient: Python driver for MySQL database.

Configure Flask-SQLAlchemy

app.py file code:

Example

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/dbname'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

code>SQLALCHEMY_DATABASE_URI: Sets the database connection URI, the format ismysql://username:password@localhost/dbname。

  • username: MySQL username.
  • password: MySQL password.
  • localhost: MySQL host address (usually local islocalhost）。
  • dbname: Database name.

Define Models and Perform Basic Operations

app.py file code:

Example

class User(db.Model):
    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}>'

@app.route('/')
def index():
    users = User.query.all()
    return '<br>'.join([f'{user.username} ({user.email})' for user in users])

if __name__ == '__main__':
    with app.app_context():
        db.create_all()  # Create database table
    app.run(debug=True)

Directly use MySQL's Python driver.

If you choose not to use SQLAlchemy, but directly use MySQL's Python driver, you can use the mysql-connector-python or PyMySQL library.

Install mysql-connector-python:

pip install mysql-connector-python

Install PyMySQL (if you choose to use PyMySQL):

pip install PyMySQL

Use mysql-connector-python to connect to MySQL

app.py file code:

Example

from flask import Flask, request, jsonify
import mysql.connector

app = Flask(__name__)

def get_db_connection():
    connection = mysql.connector.connect(
        host='localhost',
        user='username',
        password='password',
        database='dbname'
    )
    return connection

@app.route('/add_user', methods=['POST'])
def add_user():
    data = request.json
    name = data['name']
    email = data['email']

    connection = get_db_connection()
    cursor = connection.cursor()
    cursor.execute('INSERT INTO user (username, email) VALUES (%s, %s)', (name, email))
    connection.commit()
    cursor.close()
    connection.close()

    return 'User added!'

@app.route('/get_users')
def get_users():
    connection = get_db_connection()
    cursor = connection.cursor(dictionary=True)
    cursor.execute('SELECT * FROM user')
    users = cursor.fetchall()
    cursor.close()
    connection.close()

    return jsonify(users)

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

Use PyMySQL to connect to MySQL, app.py file code:

Example

from flask import Flask, request, jsonify
import pymysql

app = Flask(__name__)

def get_db_connection():
    connection = pymysql.connect(
        host='localhost',
        user='username',
        password='password',
        database='dbname',
        cursorclass=pymysql.cursors.DictCursor
    )
    return connection

@app.route('/add_user', methods=['POST'])
def add_user():
    data = request.json
    name = data['name']
    email = data['email']

    connection = get_db_connection()
    with connection.cursor() as cursor:
        cursor.execute('INSERT INTO user (username, email) VALUES (%s, %s)', (name, email))
        connection.commit()

    connection.close()
    return 'User added!'

@app.route('/get_users')
def get_users():
    connection = get_db_connection()
    with connection.cursor() as cursor:
        cursor.execute('SELECT * FROM user')
        users = cursor.fetchall()

    connection.close()
    return jsonify(users)

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