SQLAlchemy is an Object Relational Mapper. It is written in Python. SQLAlchemy is a versatile toolkit that allows developers to query and manage SQL databases efficiently. SQLAlchemy has two parts Core and ORM. This article will guide you through the Core part using the MySQL database. I am using Jupyter, you can use any editor you like.

In order to follow the article you need to install SQLAlchemy and MySQL connector. It is best practice to use virtual env to install these packages.

pip install SQLAlchemy
pip install mysql-connector-python

Database Connection

SQLAlchemy's create_engine methods return an Engine instance. The engine is the starting point in SQLAlchemy. As the engine provides information on how to talk to specific databases through DBAPI, DBAPI is mentioned in the engine connection string. The syntax for MySQL connector is mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>.

from sqlalchemy import create_engine

engine = create_engine("mysql+mysqlconnector://root:secret@localhost:3306/test")

In the above example, the engine creates a Dialect object tailored towards the MySQL database.

The dialect of the database is simply a term that defines the specific features of the SQL language that are available when accessing that database. (source)

You can test the connection using connect method. If the connection is established a Connection instance will be returned.

engine.connect()
# <sqlalchemy.engine.base.Connection at 0x24424c709a0>

If the connection is not established due to wrong credentials of configuration mysql.connector.errors.DatabaseError will be thrown. This error can be different for other DBAPIs.

If you want to use Try Except for error handling you can catch SQLAlchemy error and print the message.

from sqlalchemy import create_engine
from sqlalchemy.exc import DatabaseError

try:
    engine = create_engine("mysql+mysqlconnector://root:secret@localhost:3307/test")
    engine.connect()
except DatabaseError as e:
    print(str(e.__dict__['orig']))

Create Table

Metadata

In SQLAlchemy, metadata refers to an object that represents a collection of database objects, such as tables, indexes, and constraints. It acts as a container for these objects and provides a way to organize and manage them. The MetaData class in SQLAlchemy is used to define and store information about database structures.

from sqlalchemy import MetaData

metadata_obj = MetaData()

It acts as a central registry for tables, indexes, and other database objects, making it easier to work with the database in a structured and organized manner.

Once, all the Tables are defined, the create_all method can be called metadata_obj to create the tables in the database. It will create the table only if it does not exist.

samples = Table(
   'sample',
   metadata_obj, 
   Column('id', Integer, primary_key = True), 
   Column('some_key', String(100))
)

metadata_obj.create_all(engine)

The above code will generate a table in the database with the mentioned attributes.

Table Structure

The Table class class sqlalchemy.schema.Table in SQLAlchemy represents a table in a database.

from sqlalchemy import Table, Column, Integer, String, MetaData

metadata_obj = MetaData()

students = Table(
   'students',
   metadata_obj, 
   Column('id', Integer, primary_key = True), 
   Column('firstname', String(100)), 
   Column('lastname', String(100)), 
   Column('age', Integer)
)

SQLAlchemy has the following generic CamalCase data types to be used in the Table class. A few of these are BigInteger, Boolean, Date, Integer, String, Enum, Float, Text, and many more.

Nullable Attributes

By default all columns in a Table are nullable. A not null attribute can be set explicitly in the Column constructor.

from sqlalchemy import Table, Column, Integer, String, MetaData

metadata_obj = MetaData()

students = Table(
   'students',
   metadata_obj, 
   Column('id', Integer, primary_key = True), 
   Column('firstname', String(100), nullable=False), 
   Column('lastname', String(100), nullable=False), 
   Column('age', Integer, nullable=False)
)

If you try to create a row without any specific column like age, SQLAlchemy will throw an error.

Constraints

1. Primary Key: For single column primary key, it can be set in the Column constructor using primary_key = True.

    students = Table(
       'students',
       metadata_obj, 
       Column('id', Integer, primary_key = True), 
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
    )
   

   To set a composite primary key, the PrimaryKeyConstraint constructor can be used.

    from sqlalchemy import PrimaryKeyConstraint
   
    students = Table(
       'students',
       metadata_obj,
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
       PrimaryKeyConstraint('firstname', 'lastname', 'age', name='students_pk')
    )
   

   The above code will create a table of students with composite primary key using name and age. If try to insert two rows with the same name and age values, SQLAlchemy will throw an error.

   

2. Foreign Key: as

    from sqlalchemy import ForeignKey
   
    subject = Table(
        'subjects',
        metadata_obj,
        Column('id', Integer, primary_key=True),
        Column('title', String(100), nullable=False)
    )
   
    students = Table(
       'students',
       metadata_obj,
       Column('id', Integer, primary_key=True),
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
       Column('subject_id', Integer, ForeignKey("subjects.id"), nullable=False)
    )
   

   If we try to execute a join query to get the title of the subject and student name then we get the following.

   

   Alternatively, you can use the Column object through the c collection.

    students = Table(
       'students',
       metadata_obj,
       Column('id', Integer, primary_key=True),
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
       Column('subject_id', Integer, ForeignKey(subject.c.id), nullable=False)
    )
   

3. Unique: Unique key constraints can be set using either setting unique=True in the Column constructor.

    students = Table(
       'students',
       metadata_obj, 
       Column('id', Integer, primary_key = True), 
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
    )
   

   It can also be set using UniqueConstraint. It is a table-level construct.

    from sqlalchemy import UniqueConstraint
   
    students = Table(
       'students',
       metadata_obj, 
       Column('id', Integer, primary_key = True), 
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
       UniqueConstraint('firstname', 'lastname', 'age', name='unique_name_age')
    )
   

   Now if we try to create two rows with the same name and age.

   

   If we try to create two columns with the same email.

   

4. Check: Check constraints can be defined in the column or table level using the CheckConstraint construct.

    from sqlalchemy import PrimaryKeyConstraint, CheckConstraint
   
    students = Table(
       'students',
       metadata_obj,
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, CheckConstraint('age < 21'), nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
       PrimaryKeyConstraint('firstname', 'lastname', 'age', name='students_pk')
    )
   

   The above code implements a table that checks if the age is less than 21 or not. If the age is 21 or more database will throw an error.

   

   This can be implemented at the table level as well.

    from sqlalchemy import PrimaryKeyConstraint, CheckConstraint
   
    students = Table(
       'students',
       metadata_obj,
       Column('firstname', String(100), nullable=False), 
       Column('lastname', String(100), nullable=False), 
       Column('age', Integer, nullable=False), 
       Column('email', String(100), unique=True, nullable=False),
       CheckConstraint('age < 21', name='check_age_is_less_than_21'),
       PrimaryKeyConstraint('firstname', 'lastname', 'age', name='students_pk')
    )
   

Insert

To insert rows into database tables, SQL queries need to be executed. Insert command will be the first example of SQL query execution in this article. Data will be inserted to students table.

students = Table(
   'students',
   metadata_obj,
   Column('id', Integer, primary_key=True),
   Column('firstname', String(100), nullable=False), 
   Column('lastname', String(100), nullable=False), 
   Column('age', Integer, nullable=False), 
   Column('email', String(100), unique=True, nullable=False),
   Column('subject_id', Integer, ForeignKey(subject.c.id), nullable=False)
)

The syntax for inserting data into the table is table.insert().values(**data). So the code to run SQL insert query through SQLAlchemy is as follow.

from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.exc import DatabaseError

conn = None
try:
    engine = create_engine("mysql+mysqlconnector://root:secret@localhost:3306/test")
    conn = engine.connect()
except DatabaseError as e:
    print(str(e.__dict__['orig']))

subject = Table(
    'subjects',
    metadata_obj,
    Column('id', Integer, primary_key=True),
    Column('title', String(100), nullable=False)
)

students = Table(
   'students',
   metadata_obj,
   Column('id', Integer, primary_key=True),
   Column('firstname', String(100), nullable=False), 
   Column('lastname', String(100), nullable=False), 
   Column('age', Integer, nullable=False), 
   Column('email', String(100), unique=True, nullable=False),
   Column('subject_id', Integer, ForeignKey(subject.c.id), nullable=False)
)

ins = subject.insert().values(title = 'CSE')
conn.execute(ins)
ins = students.insert().values(firstname = 'Ritwik', lastname='Math', age=20, email="ritwikmath@gmail.com", subject_id=1)
conn.execute(ins)

Now if you want to see the params passed to the values method then use compile().params

print(ins.compile().params)
# Output
"""
{'firstname': 'Ritwik',
 'lastname': 'Math',
 'age': 20,
 'email': 'ritwikmath@gmail.com',
 'subject_id': 1}
"""

To see the raw query pass the ins variable to the str method.

str(ins)
# Output
# INSERT INTO students (firstname, lastname, age, email, subject_id) VALUES (:firstname, :lastname, :age, :email, :subject_id)

To store multiple rows in one statement, you run the following code.

conn.execute(subject.insert(), [
    {'title':'CSE'},    
    {'title':'ELECTRICAL'},
    {'title':'MECHANICAL'},
    {'title':'CHEMICAL'},
])

Another way of inserting row into table is using insert constructor.

from sqlalchemy import insert

stmt = (insert(subject).values(title='ELECTRONICS'))
conn.execute(stmt)

Select

To fetch data from a table, the select method is used.

students.select()

The above code generates a statement. This statement needs to be executed to get results.

s = students.select()
conn.execute(s)

Now this returns a cursor. To store results into a data structure that can be handled by Python various methods like fetchone and fetchall are available. fetchall method returns a python list so it is iterable.

Fetchall

s = students.select()
cursor = conn.execute(s)
print(cursor.fetchall())
# or cursor.all()

This can also be done using select construct.

from sqlalchemy import select
s = select(students)
cursor = conn.execute(s)
print(cursor.fetchall())

Fetchone

To get a single row use fetchone instead of fetchall. fetchone result is not iterable.

s = students.select().filter_by(**{'id':1})
cursor = conn.execute(s)
print(cursor.fetchone())
# Returns a LegacyRow
# (1, 'Ritwik', 'Math', 20, 'ritwikmath@gmail.com', 1)

Conditions

To introduce a condition into the query use the where or filter method.

# Using where
s = students.select().where(students.c.id == 1)
cursor = conn.execute(s)
print(cursor.fetchall())
# Using filter
s = students.select().filter(students.c.id == 1)
cursor = conn.execute(s)
print(cursor.fetchall())

If you want to use Python dict to filter results you can use filter_by. The keys in dict need to be the same as the attribute name in the table. There is no need to use c collection.

s = students.select().filter_by(**{'id':1})
cursor = conn.execute(s)
print(cursor.fetchall())
# [(1, 'Ritwik', 'Math', 20, 'ritwikmath@gmail.com', 1)]

To select specific columns mention the columns in the select method.

from sqlalchemy import select
s = select(students.c.firstname).filter_by(**{'id':1})
cursor = conn.execute(s)
print(cursor.fetchall())
# [('Ritwik',)]

First vs One

first and one both return one row.

from sqlalchemy import select
s = select(students).filter_by(**{'id':2})
cursor = conn.execute(s)
print(cursor.first())

Returns None as no row with id 2 exists.

from sqlalchemy import select
s = select(students).filter_by(**{'id':2})
cursor = conn.execute(s)
print(cursor.one())

Throws NoResultFound error.

Order by

from sqlalchemy import select
s = select(subject).order_by(subject.c.id) # Ascending
s = select(subject).order_by(subject.c.id.desc()) # Descending
cursor = conn.execute(s)
print(cursor.all())