This post is about SQLAlchemy and working with Object Relational Mappers (ORMs).

All of engines, connections, and sessions have a .execute method. The results are all the same for simple select queries, but can vary as you do more complex operations.

Table of Contents

from sqlalchemy import create_engine, inspect
from sqlalchemy.sql import text


An engine is how SQLAlchemy communicates with a database.

Creating an Engine

It’s of the form: engine = create_engine('postgresql://<user>:<password>@<hostname>:<port (optional)>/<database_name>')

It might look like this:

engine = create_engine('postgresql://julius:post123@localhost:5432/postgres')

Exploring the Database with an Engine

You can use inspect to learn more about your database.

inspector = inspect(engine)
['customers', 'spatial_ref_sys', 'geom_table']
columns = inspector.get_columns('customers')
[{'name': 'name',
  'type': TEXT(),
  'nullable': True,
  'default': None,
  'autoincrement': False,
  'comment': None},
 {'name': 'email',
  'type': TEXT(),
  'nullable': True,
  'default': None,
  'autoincrement': False,
  'comment': None},
 {'name': 'entry_num',
  'type': INTEGER(),
  'nullable': True,
  'default': None,
  'autoincrement': False,
  'comment': None},
 {'name': 'age',
  'type': INTEGER(),
  'nullable': True,
  'default': None,
  'autoincrement': False,
  'comment': None}]
[c['name'] for c in columns]
['name', 'email', 'entry_num', 'age']

Raw Queries with an Engine

Now let’s start querying

query_string = 'select * from customers'
response = engine.execute(query_string)

You get a LegacyCursorResult object as a response.

<sqlalchemy.engine.cursor.LegacyCursorResult at 0x20065e95160>
result = response.fetchall()
[('John Smith', '', 100, 25),
 ('Karen Smith', '', 101, 42),
 ('John Smith', '', 102, 25)]

You can also use a TextClause

statement = text("""select * from customers""")
<sqlalchemy.sql.elements.TextClause object at 0x0000020064993D00>

You can see what the query is by printing the statement.

select * from customers
response = engine.execute(statement)
result = response.fetchall()
[('John Smith', '', 100, 25),
 ('Karen Smith', '', 101, 42),
 ('John Smith', '', 102, 25)]


You can also use an ORM.

from geoalchemy2 import Geometry
from sqlalchemy import Column
from sqlalchemy.dialects.postgresql import INTEGER
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()

class GeomTable(Base):
    __tablename__ = "geom_table"

    id = Column("id", INTEGER, primary_key=True)
    polygon_geom = Column("polygon_geom", Geometry("POLYGON", srid=4326))
    geometry_geom = Column("geometry_geom", Geometry("GEMOETRY", srid=4326))

When you’re working with ORMs, you generally want to work with Sessions instead of engines.

Create your engine as before.

engine = create_engine('postgresql://julius:post123@localhost:5432/postgres')

Then create a Session.

Session = sessionmaker(engine)
session = Session()

Retrieving a Record

ORMs can be used to retrieve records.

response = session.query(GeomTable).filter_by(id=2)
<sqlalchemy.orm.query.Query at 0x20068f395b0>

To get the record, you could do .first(), .get(2), .all(), or .one()

.all() returns a list

[<__main__.GeomTable at 0x20068f75ca0>]

.distinct() returns another query, so you would need to call .all() on it.

<sqlalchemy.orm.query.Query at 0x20068f75a30>

.first() returns a single value (the first one), not in a list

<__main__.GeomTable at 0x20068f75ca0>
result = response.first()
<__main__.GeomTable at 0x20068f75ca0>
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState at 0x20068f75c70>,
 'geometry_geom': <WKBElement at 0x20068f75be0; 0103000020e61000000100000005000000ef30a5de4b855ec0c3308f769ab347406ed22f8149855ec0dd60a6d99ab347404ab842264a855ec035c13c23a2b34740cd16b8834c855ec01b9125c0a1b34740ef30a5de4b855ec0c3308f769ab34740>,
 'polygon_geom': <WKBElement at 0x20068f75c40; 0103000020e61000000100000005000000ef30a5de4b855ec0c3308f769ab347406ed22f8149855ec0dd60a6d99ab347404ab842264a855ec035c13c23a2b34740cd16b8834c855ec01b9125c0a1b34740ef30a5de4b855ec0c3308f769ab34740>,
 'id': 2}

There’s also a good SO post on using filter or filter_by.

Sessions and raw text

Just because you’re using a session doesn’t mean you can’t use raw text. It’s quite easy.

query_string = text('select * from geom_table where id = 3')
with Session() as session:
    record = session.query(query_string)
response = engine.execute(query_string)
result = response.fetchall()
[(3, '0103000020E610000001000000050000000000000000001040000000000000144000000000000014400000000000001440000000000000144000000000000010400000000000001040000000000000104000000000000010400000000000001440', '0103000020E610000001000000050000000000000000001040000000000000144000000000000014400000000000001440000000000000144000000000000010400000000000001040000000000000104000000000000010400000000000001440')]

Usually you’ll want it in wkt format.

query_string = 'select st_astext(polygon_geom) from geom_table where id = 3'
response = engine.execute(query_string)
result = response.fetchall()
[('POLYGON((4 5,5 5,5 4,4 4,4 5))',)]

If you want the SRID as well, you’ll need to use st_asewkt.

sess.query vs sess.execute

Creating a Record

gt = GeomTable()
gt.polygon_geom = 'POLYGON ((4 5, 5 5, 5 4, 4 4, 4 5))'
gt.geometry_geom = 'POLYGON ((4 5, 5 5, 5 4, 4 4, 4 5))'
with Session() as session:


You can also use a connection to execute a SQL query. Using a connection allows us to create transactions where either all the commands or, if there’s an error, roll it all back.

You can also make a connection:

conn = engine.connect()
trans = conn.begin()
conn.execute('INSERT INTO MY_TABLE ...)
with engine.connect() as con:

    response = con.execute(statement)
with engine.connect() as con:

    response = con.execute("select * from customers")
<sqlalchemy.engine.cursor.LegacyCursorResult at 0x20068fa2e50>
result = response.fetchall()
[('John Smith', '', 100, 25),
 ('Karen Smith', '', 101, 42),
 ('John Smith', '', 102, 25)]

Updating Databases

If you’re using an ORM and want to update an object, you’ll have to get the record, then update it and run merge.

with session(engine=engine) as sess:
    records = sess.query(MyTable).filter_by(id = 12345)
rec = records.first()
rec.my_column = 'new_value'
with session(engine=engine)) as sess:
    except Exception as ex:

Or you could pass one sess to the whole thing.

Synactic Sugar and Tricks

Get the date from a timestamp

  • date(my_timestamp)


You can learn how your query is acting by using the EXPLAIN command. For example, if you want to know what index is used in a query, you can do this:

explain select * from customers

You might get a response that looks like

Index Scan using customers_pkey on customers (cost=0.43..8.45 rows=1 width=285)

You can also use analyze. This will actually run the command and analyze the runtime statistics.

explain analyze select * from customers