One of the benefits of SQLAlchemy is that it comes with an abstraction for SQL. Let’s find out how a bit of metadata can help us write much less database code.
This post is part of my journey to learn Python. You can find the other parts of this series here. You find the code for this post in my PythonFriday repository on GitHub.
Pre-requisites
To follow along, please make sure that you have SQLAlchemy installed and that you have the Northwind database for SQLite next to your Python script. I use engine.begin() instead of engine.connect() in my examples because this way we get a transaction back that cleans up the database should an exception occur.
Create the metadata
SQLAlchemy offers the objects MetaData, Table and Column to describe the tables we want to access. I find it especially helpful that we can describe only the columns we want to use and skip the rest. That allows us to write this code snipped to access the 4 columns we care about from the Employee table:
|
1 2 3 4 5 6 7 8 9 10 |
from sqlalchemy import MetaData, Table, Column, String, Integer, Text, DateTime, Boolean, create_engine, select, insert, update, delete metadata = MetaData() employees = Table('Employee', metadata, Column('Id', Integer(), primary_key=True), Column('LastName', String(8000)), Column('FirstName', String(8000)), Column('BirthDate', String(8000)) ) |
We can ask the metadata variable what it knows about our database with this snipped:
|
1 2 3 4 5 |
def show_metadata(): for t in metadata.sorted_tables: print(f"Table {t.name}:") for c in t.columns: print(f"{c} ({c.type})") |
Table Employee:
Employee.Id (INTEGER)
Employee.LastName (VARCHAR(8000))
Employee.FirstName (VARCHAR(8000))
Employee.BirthDate (VARCHAR(8000))
Create (or insert)
With the metadata in place, our insert code works with Python objects and no longer needs any raw SQL statements:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
def do_insert(): stmt = insert(employees).values( LastName='Collins', FirstName='Arnold', BirthDate='2000-01-31') new_id = 0 connection_string = "sqlite:///Northwind_small.sqlite" engine = create_engine(connection_string, echo=False) with engine.begin() as con: result = con.execute(stmt) new_id = result.inserted_primary_key['Id'] print(f"New Id: {new_id}") return new_id |
First we create in insert() object that we connect to the metadata object for our table. On this object we set the values we want to insert as named arguments. Make sure that you write the column names exactly the same way as you used in the metadata.
To execute our Insert object, we need an engine and open up a transaction as we did in the raw SQL example. An important difference is that we now get a result back that contains in the attribute inserted_primary_key the value for the primary key of our newly created row.
Read (or select)
We can select an employee by its Id with this code:
|
1 2 3 4 5 6 7 8 9 10 11 |
def select_by_id(id): stmt = select(employees).where(employees.c.Id == id) connection_string = "sqlite:///Northwind_small.sqlite" engine = create_engine(connection_string, echo=False) with engine.begin() as con: result = con.execute(stmt).first() if result: print(result) else: print(f"no rows found with Id == {id}") |
We create a select() object and use the .where() method to specify which rows we want. In the example above I use employees.c.Id to tell SQLAlchemy that I want to match my parameter to the Id column (inside .c on the metadata object for your table is a list of all columns). Next week we take a deeper look at the different ways we can select rows in SQLAlchemy Core.
Since we search by the primary key, there is only one row that matches our where() and we can use the .first() method to get back the expected row (instead of a result set that we need to iterate over). Be aware that first() can return None if no row matches our select statement. Make sure that you put a check in place before you access values for an object that does not exist.
Update
We can update an employee with this code:
|
1 2 3 4 5 6 7 8 9 |
def do_update(id): stmt = update(employees).values( FirstName="Michael" ).where(employees.c.Id == id) connection_string = "sqlite:///Northwind_small.sqlite" engine = create_engine(connection_string, echo=False) with engine.begin() as con: con.execute(stmt) |
Our update() object gets the new values as named arguments and a .where() clause. If you do not specify which rows to update, SQLAlchemy will update every row in your database.
Delete
We can delete an employee with this code:
|
1 2 3 4 5 6 7 |
def do_delete(id): stmt = delete(employees).where(employees.c.Id == id) connection_string = "sqlite:///Northwind_small.sqlite" engine = create_engine(connection_string, echo=False) with engine.begin() as con: con.execute(stmt) |
As with the update() object, make sure that your delete() object always gets a .where() clause. Otherwise you delete all rows in your table.
Next
SQLAlchemy Core with metadata lets us access our database without the need to write SQL statements. We still need to know the operations and define a where clause, but the code we write is on a much higher abstraction level. Next week we look at filters and how we can combine them to get back exactly the rows we want.