SQLAlchemy
Data Science @ Statnett
Simeon Simeonov
Agenda
- SQLAlchemy - Design & overview
- SQLAlchemy - A small practical example
- Q & A
What is SQLAlchemy?
SQLAlchemy is a Python library created by Mike Bayer to provide a high-level Pythonic interface to RDBMS such as PostgreSQL, SQLite, MySQL, Oracle, DB2.
SQLAlchemy includes RDBMS-independent SQL expression language and an object-relational mapper (ORM).
Why use SQLAlchemy?
free software - free as in "freedom" (MIT licensed)
portability - the programming interface is independent of the type of RDBMS and connector used
security - no more SQL injections
abstraction - no need to bother with complex JOINs
object-orientation - you work with objects instead of tables and rows
performance - exploits the likehood of reusing a particular query
flexibility - you can override almost anything
0.0125 karma points for each MB transferred (2021)
Basic architecture
SQLAlchemy consists of several components, including the ORM.
- Engine- manages the connection pool and the RDBMS-independent SQL dialect layer
- MetaData - used to collect and organize information about your table layout (schema)
- SQL expression language - provides an API to execute your queries and updates against your tables, all from Python, and all in a database-independent way (low-level interface)
- ORM - provides a convenient way to add database persistence to your Python objects without requiring you to design your objects around the database, or the database around the objects (high-level interface)
- Session - establishes all conversations with the RDBMS and represents a "holding zone" for all the objects which you've loaded or associated with it during its lifespan
Example
SQLAlchemy gives us the choice between classical mapping and the newer declarative mapping
import sqlalchemy
sqlalchemy.__version__
# Out: '1.3.23'
from sqlalchemy import create_engine
# engine = create_engine("postgresql+psycopg2://user:zipassword@localhost/mydb" , echo=True)
# The string form of the URL is dialect+driver://user:password@host/dbname[?key=value..],
# where dialect is a database name such as mysql, oracle, postgresql, etc.,
# and driver the name of a DBAPI, such as psycopg2, pyodbc, cx_oracle
# The echo flag is a shortcut to setting up SQLAlchemy logging,
# which is accomplished via Python’s standard logging module.
# engine = create_engine("sqlite:///library.db", echo=True)
engine = create_engine("sqlite:///:memory:", echo=True)
from sqlalchemy import Column, ForeignKey, Integer, String, Table
metadata = MetaData()
authors_table = Table(
"authors",
metadata,
Column("author_id", Integer, primary_key=True),
Column("name", String),
) # Column("name", String(50)) is possible
books_table = Table(
"books",
metadata,
Column("book_id", Integer, primary_key=True),
Column("title", String),
Column("description", String),
Column("author_id", ForeignKey('authors.author_id')),
)
metadata.create_all(engine) # creates the tables
Example (cont...)
Use of SQL expression language
insert_stmt = authors_table.insert(bind=engine)
type(insert_stmt)
# Out: <class 'sqlalchemy.sql.expression.Insert'>
print(insert_stmt)
# Out: INSERT INTO authors (id, name) VALUES (:id,:name)
compiled_stmt = insert_stmt.compile()
print(compiled_stmt.params)
# Out: {'id': None, 'name': None}
insert_stmt.execute(name="Alexandre Dumas") # insert a single entry
insert_stmt.execute([{"name": "Mr X"}, {"name": "Mr Y"}]) # a list of entries
metadata.bind = engine # no need to explicitly bind the engine from now on
select_stmt = authors_table.select(authors_table.c.id==2)
result = select_stmt.execute()
result.fetchall()
# Out: [(1, 'Mr X')]
del_stmt = authors_table.delete()
del_stmt.execute(whereclause=text("name='Mr Y'"))
del_stmt.execute() # delete all
Example (cont...)
Use of classical mapping
from sqlalchemy.orm import backref, mapper, relation
class Author:
def __init__(self, name):
self.name = name
def __str__(self):
return self.name
class Book:
def __init__(self, title, description, author):
self.title = title
self.description = description
self.author = author
def __str__(self):
return self.title
mapper(Book, books_table)
mapper(Author, authors_table, properties = {"books": relation(Book, backref="author")})
Example (cont...)
Doing the same thing the easy way with declarative mapping
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref
Base = declarative_base()
class Author(Base):
__tablename__ = "authors"
author_id = Column(Integer, primary_key=True)
name = Column(String)
def __init__(self, name):
self.name = name
def __str__(self):
return self.name
class Book(Base):
__tablename__ = "books" # self.__table__ will be available for our objects
book_id = Column(Integer, primary_key=True)
title = Column(String)
description = Column(String)
author_id = Column(Integer, ForeignKey("authors.author_id"))
author = relationship(Author, backref=backref("books", order_by=title))
def __init__(self, title, description, author):
self.title = title
self.description = description
self.author = author
def __str__(self):
return self.title
Base.metadata.create_all(engine) # create tables
Example (cont...)
Creating instances
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine) # bound session
session = Session()
author_1 = Author("Richard Dawkins")
author_2 = Author("Matt Ridley")
book_1 = Book("The Red Queen", "A popular science book", author_2)
book_2 = Book("The Selfish Gene", "A popular science book", author_1)
book_3 = Book("The Blind Watchmaker", "The theory of evolutio", author_1) # typo
session.add(author_1)
session.add(author_2)
session.add(book_1)
session.add(book_2)
session.add(book_3)
# or simply session.add_all([author_1, author_2, book_1, book_2, book_3])
# session.flush()
session.commit() # flushes (issues the statements and sends them to the RDBMS) and commits
book_3.description = "The theory of evolution" # update the object
book_3 in session # check whether the object is in the session
# Out: True
session.commit()
Example (cont...)
Queries
session.query(Book).order_by(Book.book_id) # returns a Query instance with a .statement attribute
session.query(Book).order_by(Book.book_id).all() # returns an object-list
# return all book objects where title == "The Selfish Gene"
session.query(Book).filter(Book.title == "The Selfish Gene").order_by(Book.book_id).all()
# using LIKE
session.query(Book).filter(Book.title.like("The%")).order_by(Book.book_id).all()
query = session.query(Book).filter(Book.book_id == 9).order_by(Book.book_id)
query.count() # returns 0
query.all() # returns an empty list
query.first() # returns None
query.one() # raises NoResultFound exception
query = session.query(Book).filter(Book.book_id == 1).order_by(Book.book_id)
book_1 = query.one()
book_1.description # returns "A popular science book"
book_1.author.books # returns a list of Book-objects representing all the books from the same author.
# get a list of all Book-instances where the author"s name is "Richard Dawkins"
session.query(Book).filter(Book.author_id == Author.author_id).filter(Author.name == "Richard Dawkins").all()
session.query(Book).join(Author).filter(Author.name == "Richard Dawkins").all()
session.query(Book).\
from_statement("SELECT b.* FROM books b, authors a WHERE b.author_id = a.author_id AND a.name=:name").\
params(name="Richard Dawkins").all()
session.query(Book).filter(Book.author == author_1).all()
... there is more???
Some nice features
import pandas as pd
from sqlalchemy import func
class Book(Base):
# ...
author = relationship(
Author, backref=backref("books", lazy="dynamic", order_by=title)
)
# ...
@hybrid_property
def newly_arrived(self):
return self.book_id > 2
@newly_arrived.expression
def newly_arrived(cls):
return cls.book_id > 2
# return func.abs(cls.book_id) > 2
# .books is now a Query object
query = author_obj.books.filter(Book.title.ilike("%red%"))
session.query(Book).filter(Book.newly_arrived.is_(True)).all()
# Out: [<__main__.Book at 0x7f132bdf0130>]
# WHERE (abs(books.book_id) > ?) IS 1 ... in the case of func.abs
# with Pandas
df = pd.read_sql_table("my_table", con=session.get_bind()) # or con=engine
df = pd.read_sql_query(query.statement, engine)