SQL Alchemy for Beginners¶

An introduction to the core features of SQL Alchemy for Object Relational Mapping (ORM) in Python, together with Alembic for managing migrations.

Initialise Alembic¶

In [1]:

!alembic init migrations

!alembic init migrations

  Creating directory
  /Users/alexioannides/Dropbox/data_science/workspace/python/data-science-and-ml-
  notebook/sqlalchemy/migrations ...  done
  Creating directory
  /Users/alexioannides/Dropbox/data_science/workspace/python/data-science-and-ml-
  notebook/sqlalchemy/migrations/versions ...  done
  Generating /Users/alexioannides/Dropbox/data_science/workspace/python/data-
  science-and-ml-notebook/sqlalchemy/migrations/script.py.mako ...  done
  Generating /Users/alexioannides/Dropbox/data_science/workspace/python/data-
  science-and-ml-notebook/sqlalchemy/migrations/env.py ...  done
  Generating /Users/alexioannides/Dropbox/data_science/workspace/python/data-
  science-and-ml-notebook/sqlalchemy/migrations/README ...  done
  File /Users/alexioannides/Dropbox/data_science/workspace/python/data-science-
  and-ml-notebook/sqlalchemy/alembic.ini already exists, skipping
  Please edit configuration/connection/logging settings in
  '/Users/alexioannides/Dropbox/data_science/workspace/python/data-science-and-ml-
  notebook/sqlalchemy/alembic.ini' before proceeding.

Change the following in alembic.ini before proceeding:

sqlalchemy.url = sqlite:///data.db

Define Models using Classes¶

These are contained in models.py - they need to be imported by migrations/env.py, so they can't be defined within a Jupyter notebook. We reproduce the contents of models.py here for convenience.

"""
All data Models can be found here.
"""
from typing import Any, Dict

from sqlalchemy import (
    Column,
    Float,
    Integer,
    String,
    ForeignKey
)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()


class Person(Base):
    """Personal information."""
    
    __tablename__ = "person"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    address_id = Column(Integer, ForeignKey("address.id"))
    name = Column(String, nullable=False)
    age = Column(Float, nullable=False)
    
    address = relationship("Address", back_populates="person")

    def dict(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "address_id": self.address_id,
            "name": self.name,
            "age": self.age
        }


class Address(Base):
    """Address information."""
    
    __tablename__ = "address"
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    street = Column(String, nullable=False)
    city = Column(String, nullable=False)
    postcode = Column(String, nullable=False)

    person = relationship("Person", back_populates="address")
    
    def dict(self) -> Dict[str, Any]:
        return {
            "id": self.id,
            "street": self.street,
            "city": self.city,
            "postcode": self.postcode
        }

Apply Models to the Database¶

Modify migrations/env.py as follows,

# ...

# add your model's MetaData object here
# for 'autogenerate' support
from models import Base
target_metadata = Base.metadata

# ...

And then we're ready for Alembic to setup the database schema for us. We start by creating the migrations.

In [2]:

!alembic revision --autogenerate -m "Initial Migration"

!alembic revision --autogenerate -m "Initial Migration"

INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'address'
INFO  [alembic.autogenerate.compare] Detected added table 'person'
  Generating /Users/alexioannides/Dropbox/data_science/workspace/python/data-
  science-and-ml-
  notebook/sqlalchemy/migrations/versions/c31efd831ee7_initial_migration.py ...  done

And then we apply them to the DB.

In [3]:

!alembic upgrade head

!alembic upgrade head

INFO  [alembic.runtime.migration] Context impl SQLiteImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> c31efd831ee7, Initial Migration

Interacting with Data via SQL Alchemy Models¶

Imports¶

In [4]:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from models import Person, Address

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models import Person, Address

Session¶

Create a session factory.

In [5]:

session_factory = sessionmaker(bind=create_engine("sqlite:///data.db", echo=False))
session = session_factory()

session_factory = sessionmaker(bind=create_engine("sqlite:///data.db", echo=False)) session = session_factory()

Create some addresses and commit them.

In [6]:

session.add_all(
    [Address(street="40 Guernsey Grove", city="London", postcode="SE24 9DE"),
     Address(street="42 Guernsey Grove", city="London", postcode="SE24 9DE")]
)
session.commit()

session.add_all( [Address(street="40 Guernsey Grove", city="London", postcode="SE24 9DE"), Address(street="42 Guernsey Grove", city="London", postcode="SE24 9DE")] ) session.commit()

Let's run a quick query to make sure they've persisted.

In [7]:

addresses = session.query(Address).all()
for address in addresses:
    print(f"[id={address.id}]: {address.street}, {address.city}, {address.postcode}")

addresses = session.query(Address).all() for address in addresses: print(f"[id={address.id}]: {address.street}, {address.city}, {address.postcode}")

[id=1]: 40 Guernsey Grove, London, SE24 9DE
[id=2]: 42 Guernsey Grove, London, SE24 9DE

Now create some people.

In [8]:

session.add_all(
    [Person(name="Alex Ioannides", age="41", address_id=1),
     Person(name="Sue Mortimer", age="53", address_id=2)]
)
session.commit()

session.add_all( [Person(name="Alex Ioannides", age="41", address_id=1), Person(name="Sue Mortimer", age="53", address_id=2)] ) session.commit()

And make sure that they've persisted correctly.

In [9]:

people = session.query(Person).all()
for person in people:
    print(
        f"[id={person.id}]: {person.name} ({person.age}) @ {person.address.street}, "
        f"{person.address.city}, {person.address.postcode}"
    )

people = session.query(Person).all() for person in people: print( f"[id={person.id}]: {person.name} ({person.age}) @ {person.address.street}, " f"{person.address.city}, {person.address.postcode}" )

[id=1]: Alex Ioannides (41.0) @ 40 Guernsey Grove, London, SE24 9DE
[id=2]: Sue Mortimer (53.0) @ 42 Guernsey Grove, London, SE24 9DE

Extracting the same information via a JOIN.

In [10]:

people = session.query(Person).join(Address, Person.address_id==Address.id).all()
for person in people:
    print(
        f"[id={person.id}]: {person.name} ({person.age}) @ {address.street}, "
        f"{address.city}, {address.postcode}"
    )

people = session.query(Person).join(Address, Person.address_id==Address.id).all() for person in people: print( f"[id={person.id}]: {person.name} ({person.age}) @ {address.street}, " f"{address.city}, {address.postcode}" )

[id=1]: Alex Ioannides (41.0) @ 42 Guernsey Grove, London, SE24 9DE
[id=2]: Sue Mortimer (53.0) @ 42 Guernsey Grove, London, SE24 9DE

Or via implicit join.

In [11]:

query = session.query(Person, Address).filter(Person.address_id==Address.id).all()
for person, address in query:
    print(
        f"[id={person.id}]: {person.name} ({person.age}) @ {address.street}, "
        f"{address.city}, {address.postcode}"
    )

query = session.query(Person, Address).filter(Person.address_id==Address.id).all() for person, address in query: print( f"[id={person.id}]: {person.name} ({person.age}) @ {address.street}, " f"{address.city}, {address.postcode}" )

[id=1]: Alex Ioannides (41.0) @ 40 Guernsey Grove, London, SE24 9DE
[id=2]: Sue Mortimer (53.0) @ 42 Guernsey Grove, London, SE24 9DE