SQLAlchemy
Data Engineering @ 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
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
# option 1: classical mapping
# explicitly defining Table objects and mapping them to pure Python base classes
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..],
# engine = create_engine("sqlite:///library.db", echo=True)
engine = create_engine("sqlite:///:memory:", echo=True)
from sqlalchemy import Column, MetaData, Table
from sqlalchemy import DateTime, ForeignKey, Integer, Numeric, String
metadata = MetaData()
production_types_table = Table(
"production_types",
metadata,
Column("production_type_id", Integer, primary_key=True),
Column("code", String(3), nullable=False, unique=True),
Column("description", String), # Column("name", String(128)) is possible
)
bidding_areas_table = Table(
"bidding_areas",
metadata,
Column("bidding_area_id", Integer, primary_key=True),
Column("code", String(3), nullable=False, unique=True),
Column("name", String(32)),
)
production_plans_table = Table(
"production_plans",
metadata,
Column("record_created_time", DateTime(timezone=False), primary_key=True),
Column("start_time", DateTime(timezone=False), primary_key=True),
Column("bidding_area_id", Integer, ForeignKey("bidding_areas.bidding_area_id"), primary_key=True),
Column("production_type_id", Integer, ForeignKey("production_types.production_type_id"), primary_key=True),
Column("value", Numeric, nullable=False),
)
metadata.create_all(engine) # creates the tables
Example (cont...)
Use of SQL expression language
# option 1: classical mapping (continues)
# Using the SQL expression language (low level interface)
from sqlalchemy import text
insert_stmt = bidding_areas_table.insert(bind=engine)
type(insert_stmt)
# Out: <class 'sqlalchemy.sql.dml.Insert'>
print(insert_stmt)
# Out: INSERT INTO bidding_areas (bidding_area_id, code, name) VALUES (?, ?, ?)
compiled_stmt = insert_stmt.compile()
print(compiled_stmt.params)
# Out: {'bidding_area_id': None, 'code': None, 'name': None}
insert_stmt.execute(bidding_area_id=1, code="NO1", name="Elspot NO1") # insert a single entry
# ... or a list of entries
insert_stmt.execute(
[
{"bidding_area_id": 2, "code": "NO2", "name": "Elspot NO2"},
{"bidding_area_id": 3, "code": "NO3", "name": "Elspot NO3"},
{"bidding_area_id": 4, "code": "NO4", "name": "Elspot NO4"},
{"bidding_area_id": 5, "code": "NO5", "name": "Elspot NO5"},
{"bidding_area_id": 6, "code": "NO6", "name": "Elspot NO6"},
]
)
metadata.bind = engine # no need to explicitly bind the engine from now on
select_stmt = bidding_areas_table.select(bidding_areas_table.c.bidding_area_id==2)
result = select_stmt.execute()
result.fetchall()
# Out: [(2, 'NO2', 'Elspot NO2')]
del_stmt = bidding_areas_table.delete()
del_stmt.execute(whereclause=text("name='Elspot NO6'"))
del_stmt.execute() # delete NO6
Example (cont...)
Use of classical mapping
# option 1: classical mapping (continues)
# Defining regular base classes and mapping them to the Table objects
from sqlalchemy.orm import mapper
class ProductionType:
def __init__(self, code, description):
self.code = code
self.description = description
def __str__(self):
return self.code
class BiddingArea:
def __init__(self, code, name):
self.code = code
self.name = name
def __str__(self):
return self.code
mapper(ProductionType, production_types_table)
mapper(BiddingArea, bidding_areas_table)
Example (cont...)
Use of classical mapping
from sqlalchemy.orm import relationship
class ProductionPlan:
def __init__(self, record_created_time, start_time, production_type, bidding_area, value):
self.record_created_time = record_created_time
self.start_time = start_time
self.production_type = production_type
self.bidding_area = bidding_area
self.value = value
def __str__(self):
return (
f"{self.record_created_time} {self.start_time} "
f"{self.production_type} {self.bidding_area} {self.value}"
)
mapper(
ProductionPlan,
production_plans_table,
properties = {
"production_type": relationship(ProductionType, backref="production_plans"),
"bidding_area": relationship(BiddingArea, backref="production_plans"),
},
)
Example (cont...)
Doing the same thing the easy way with declarative mapping
# option 2: declarative mapping
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
class ProductionType(Base):
__tablename__ = "production_types"
production_type_id = Column(Integer, primary_key=True)
code = Column(String(3), nullable=False, unique=True)
description = Column(String)
def __init__(self, code, description):
self.code = code
self.description = description
def __str__(self):
return self.code
class BiddingArea(Base):
__tablename__ = "bidding_areas"
bidding_area_id = Column(Integer, primary_key=True)
code = Column(String(3), nullable=False, unique=True)
name = Column(String(32))
def __init__(self, code, name):
self.code = code
self.name = name
def __str__(self):
return self.code
Example (cont...)
Doing the same thing the easy way with declarative mapping
# option 2: declarative mapping (continues)
from sqlalchemy.orm import relationship, backref
class ProductionPlan(Base):
__tablename__ = "production_plans"
record_created_time = Column(DateTime(timezone=False), primary_key=True)
start_time = Column(DateTime(timezone=False), primary_key=True)
bidding_area_id = Column(Integer, ForeignKey("bidding_areas.bidding_area_id"), primary_key=True)
production_type_id = Column(Integer, ForeignKey("production_types.production_type_id"), primary_key=True)
value = Column(Numeric, nullable=False)
# defining relationships.
# the defined attributes will reference 'ProductionType' and 'BiddingArea' objects
production_type = relationship(ProductionType, backref=backref("production_plans"))
bidding_area = relationship(BiddingArea, backref=backref("production_plans"))
def __init__(self, record_created_time, start_time, production_type, bidding_area, value):
self.record_created_time = record_created_time
self.start_time = start_time
self.production_type = production_type # a 'ProductionType' object
self.bidding_area = bidding_area # a 'BiddingArea' object
self.value = value
def __str__(self):
return (
f"{self.record_created_time} {self.start_time} "
f"{self.production_type} {self.bidding_area} {self.value}"
)
Base.metadata.create_all(engine) # create tables
Example (cont...)
Creating instances
# adding some data...
import datetime
import decimal
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine) # bound session
session = Session()
bidding_area1 = BiddingArea("NO1", "Elspot NO1")
session.add(bidding_area1)
session.add_all(
[
BiddingArea("NO2", "Elspot NO2"),
BiddingArea("NO3", "Elspot NO3"),
BiddingArea("NO4", "Elspot NO4"),
BiddingArea("NO5", "Elspot NO5"),
]
)
production_type_B37 = ProductionType("B37", "Thermal unspecified")
production_type_B30 = ProductionType("B30", "Wind unspecified")
session.add_all(
[
ProductionType("B19", "Wind Onshore"),
ProductionType("B10", "Hydro-electric pure pumped storage head installation"),
ProductionType("B11", "Hydro Run-of-river head installation"),
ProductionType("B12", "Hydro-electric storage head installation"),
ProductionType("A04", "Generation"),
production_type_B37,
production_type_B30,
]
)
Example (cont...)
Creating instances
# adding some production plans...
session.add(
ProductionPlan(
datetime.datetime.now(),
datetime.datetime(2022, 11, 2, 1, 0),
production_type_B37,
bidding_area1,
decimal.Decimal("80.5"),
)
)
production_plan2 = ProductionPlan(
datetime.datetime.now(),
datetime.datetime(2022, 11, 2, 2, 0),
production_type_B37,
bidding_area1,
decimal.Decimal("90.5"),
)
session.add(production_plan2)
session.flush() # execute pending operations
session.commit() # execute and commit pending operations
production_plan2.value = decimal.Decimal("70.5")
production_plan2 in session
# Out: True
session.commit()
Example (cont...)
Queries
import pandas as pd
session.query(ProductionPlan).order_by(ProductionPlan.start_time) # returns a Query instance
session.query(ProductionPlan).order_by(ProductionPlan.start_time).all() # returns an object-list
# return all production plans where start_time after 2022-09-01 00:00
session.query(ProductionPlan).filter(ProductionPlan.start_time > datetime.datetime(2022, 9, 1, 0, 0)).all()
# return production plans with value > 80
query = session.query(ProductionPlan).filter(ProductionPlan.value > 80).order_by(ProductionPlan.start_time)
query.count() # returns 1
production_plan = query.first() # returns the first object (element)
production_plan = query.one() # raises NoResultFound exception or MultipleResultsFound in case elements != 1
# generate Pandas DataFrame from a query or entire table
df = pd.read_sql_table("my_table", con=session.get_bind()) # or con=engine
df = pd.read_sql_query(query.statement, engine)
# return production plans with production type 'B37'
session.query(ProductionPlan).filter(
ProductionPlan.production_type_id == ProductionType.production_type_id
).filter(ProductionType.code == "B37").all()
session.query(ProductionPlan).join(ProductionType).filter(ProductionType.code == "B37").all()
session.query(ProductionPlan).filter(ProductionPlan.production_type == production_type_B37).all()
session.query(
ProductionPlan
).from_statement(
text(
"SELECT pp.* FROM production_plans pp, production_types pt "
"WHERE pp.production_type_id = pt.production_type_id AND pt.code=:code"
)
).params(code="B37").all()