Overview
The MySQL dialect provides support for both MySQL and MariaDB databases. While MariaDB maintains fundamental compatibility with MySQL’s protocols, SQLAlchemy can detect and accommodate differences between the two databases.
Version Support
Normal support: MySQL 5.6+ / MariaDB 10+
Best effort: MySQL 5.0.2+ / MariaDB 5.0.2+
Installation
mysqlclient (Default)
PyMySQL
MySQL Connector
asyncmy (Async)
aiomysql (Async)
Maintained fork of MySQL-Python. Very stable, supports Python 2 and 3. Pure Python driver. 100% compatible with MySQL-Python. pip install mysql-connector-python
Official MySQL driver from Oracle. Async driver based on PyMySQL.
Connecting to MySQL
Connection Strings
mysqlclient (sync)
PyMySQL
MySQL Connector
asyncmy (async)
aiomysql (async)
from sqlalchemy import create_engine
engine = create_engine(
"mysql+mysqldb://user:password@localhost:3306/mydb?charset=utf8mb4"
)
Default Driver
If no driver is specified, mysqldb (mysqlclient) is used:
engine = create_engine(
"mysql://user:password@localhost/mydb?charset=utf8mb4"
)
Always specify charset=utf8mb4 in your connection string for proper Unicode support.
MariaDB Support
Automatic Detection
SQLAlchemy automatically detects MariaDB and adjusts behavior:
engine = create_engine(
"mysql+pymysql://user:pass@mariadb_host/dbname?charset=utf8mb4"
)
# SQLAlchemy detects MariaDB automatically
MariaDB-Only Mode
Enforce MariaDB-only connections:
engine = create_engine(
"mariadb+pymysql://user:pass@mariadb_host/dbname?charset=utf8mb4"
)
In MariaDB-only mode, replace mysql_ prefixes with mariadb_ in table options:
mysql_engine → mariadb_engine
mysql_charset → mariadb_charset
from sqlalchemy import Table, Column, Integer, String, MetaData
metadata = MetaData()
my_table = Table(
'mytable' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'data' , String( 50 )),
# Use both for compatibility
mysql_engine = 'InnoDB' ,
mariadb_engine = 'InnoDB'
)
MySQL-Specific Features
Storage Engines
Specify storage engine for tables:
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'name' , String( 50 )),
mysql_engine = 'InnoDB' , # or 'MyISAM', 'Memory', etc.
mysql_charset = 'utf8mb4'
)
ENUM Type
MySQL’s native ENUM type:
from sqlalchemy.dialects.mysql import ENUM
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column(
'status' ,
ENUM( 'pending' , 'active' , 'inactive' , 'deleted' )
)
)
SET Type
MySQL’s SET type for multiple values:
from sqlalchemy.dialects.mysql import SET
permissions = Table(
'permissions' ,
metadata,
Column( 'user_id' , Integer, primary_key = True ),
Column(
'roles' ,
SET( 'read' , 'write' , 'admin' , 'delete' )
)
)
JSON Type
MySQL 5.7+ supports native JSON:
from sqlalchemy.dialects.mysql import JSON
products = Table(
'products' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'attributes' , JSON )
)
# Query with JSON path
stmt = products.select().where(
products.c.attributes[ 'price' ].astext.cast(Integer) > 100
)
MySQL Numeric Types
from sqlalchemy.dialects.mysql import (
TINYINT , # 1 byte integer
SMALLINT , # 2 byte integer
MEDIUMINT , # 3 byte integer
INTEGER , # 4 byte integer
BIGINT , # 8 byte integer
DECIMAL , # Fixed precision
FLOAT , # Floating point
DOUBLE , # Double precision
REAL # Alias for DOUBLE
)
scores = Table(
'scores' ,
metadata,
Column( 'id' , INTEGER( unsigned = True ), primary_key = True ),
Column( 'value' , TINYINT( unsigned = True ))
)
MySQL String Types
from sqlalchemy.dialects.mysql import (
CHAR ,
VARCHAR ,
TINYTEXT ,
TEXT ,
MEDIUMTEXT ,
LONGTEXT
)
articles = Table(
'articles' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'title' , VARCHAR( 255 )),
Column( 'summary' , TEXT ),
Column( 'content' , LONGTEXT )
)
MySQL Binary Types
from sqlalchemy.dialects.mysql import (
BINARY ,
VARBINARY ,
TINYBLOB ,
BLOB ,
MEDIUMBLOB ,
LONGBLOB
)
files = Table(
'files' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'thumbnail' , BLOB ),
Column( 'data' , LONGBLOB )
)
Other MySQL Types
from sqlalchemy.dialects.mysql import (
BIT , # Bit field
YEAR , # Year (1901-2155)
TIME , # Time
DATETIME , # Date and time
TIMESTAMP , # Timestamp
BOOLEAN # Alias for TINYINT(1)
)
MariaDB-Specific Types
from sqlalchemy.dialects.mysql import INET4 , INET6
network = Table(
'network' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'ipv4' , INET4 ),
Column( 'ipv6' , INET6 )
)
INSERT … ON DUPLICATE KEY UPDATE
from sqlalchemy.dialects.mysql import insert
stmt = insert(users).values(
id = 1 ,
name = 'Alice' ,
email = 'alice@example.com'
)
# Update on duplicate key
stmt = stmt.on_duplicate_key_update(
name = stmt.inserted.name,
email = stmt.inserted.email
)
connection.execute(stmt)
Full-Text Search
from sqlalchemy.dialects.mysql.expression import match
articles = Table(
'articles' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'title' , String( 200 )),
Column( 'content' , TEXT )
)
# Full-text search
stmt = articles.select().where(
match(articles.c.title, articles.c.content, against = 'python database' )
)
Create FULLTEXT index:
from sqlalchemy import Index
Index(
'ft_articles' ,
articles.c.title,
articles.c.content,
mysql_prefix = 'FULLTEXT'
)
Connection Timeouts
MySQL closes idle connections after 8 hours by default. Use pool_recycle:
engine = create_engine(
"mysql+pymysql://user:pass@localhost/mydb" ,
pool_recycle = 3600 , # Recycle connections after 1 hour
pool_pre_ping = True # Verify connections before using
)
SSL Connections
Query String
connect_args
engine = create_engine(
"mysql+pymysql://user:pass@host/db"
"?ssl_ca=/path/to/ca.pem"
"&ssl_cert=/path/to/client-cert.pem"
"&ssl_key=/path/to/client-key.pem"
)
engine = create_engine(
"mysql+pymysql://user:pass@host/db" ,
connect_args = {
"ssl" : {
"ca" : "/path/to/ca.pem" ,
"cert" : "/path/to/client-cert.pem" ,
"key" : "/path/to/client-key.pem"
}
}
)
Case Sensitivity
MySQL table and column names may be case-sensitive depending on the OS:
Linux : Case-sensitive by default
Windows/macOS : Case-insensitive
For portability, always use lowercase table and column names.
AUTO_INCREMENT
MySQL uses AUTO_INCREMENT for primary keys:
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ), # AUTO_INCREMENT implied
Column( 'name' , String( 50 ))
)
Customize starting value:
users = Table(
'users' ,
metadata,
Column( 'id' , Integer, primary_key = True ),
Column( 'name' , String( 50 )),
mysql_auto_increment = 1000
)
LIMIT with Offset
from sqlalchemy.dialects.mysql import limit
# Standard approach
stmt = users.select().limit( 10 ).offset( 20 )
# MySQL-specific LIMIT clause
from sqlalchemy import select
stmt = select(users).limit( 10 , 20 ) # LIMIT 20, 10
Use InnoDB for transactional tables with foreign keys
Always specify charset=utf8mb4 for full Unicode support
Use connection pooling with pool_pre_ping=True
Set pool_recycle to prevent connection timeouts
Create proper indexes on frequently queried columns
Use EXPLAIN to analyze query performance
Common Issues
Character Encoding
Always use utf8mb4 for full Unicode support including emojis:
engine = create_engine(
"mysql+pymysql://user:pass@localhost/mydb?charset=utf8mb4" ,
connect_args = { "charset" : "utf8mb4" }
)
Connection Lost
Use pool_pre_ping to check connections:
engine = create_engine(
"mysql://..." ,
pool_pre_ping = True ,
pool_recycle = 3600
)
See Also
Dialect Overview Learn about SQLAlchemy’s dialect system
MySQL Documentation Official MySQL documentation