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
Engines
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)
inspector.get_table_names()
['customers', 'spatial_ref_sys', 'geom_table']
columns = inspector.get_columns('customers')
columns
[{'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.
response
<sqlalchemy.engine.cursor.LegacyCursorResult at 0x20065e95160>
result = response.fetchall()
result
[('John Smith', 'js@gmail.com', 100, 25),
('Karen Smith', 'ks@gmail.com', 101, 42),
('John Smith', 'js@gmail.com', 102, 25)]
You can also use a TextClause
statement = text("""select * from customers""")
statement
<sqlalchemy.sql.elements.TextClause object at 0x0000020064993D00>
You can see what the query is by printing the statement.
print(statement)
select * from customers
response = engine.execute(statement)
result = response.fetchall()
result
[('John Smith', 'js@gmail.com', 100, 25),
('Karen Smith', 'ks@gmail.com', 101, 42),
('John Smith', 'js@gmail.com', 102, 25)]
ORMs
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)
response
<sqlalchemy.orm.query.Query at 0x20068f395b0>
To get the record, you could do .first(), .get(2), .all(), or .one()
.all() returns a list
response.all()
[<__main__.GeomTable at 0x20068f75ca0>]
.distinct() returns another query, so you would need to call .all() on it.
response.distinct()
<sqlalchemy.orm.query.Query at 0x20068f75a30>
.first() returns a single value (the first one), not in a list
response.first()
<__main__.GeomTable at 0x20068f75ca0>
result = response.first()
result
<__main__.GeomTable at 0x20068f75ca0>
result.__dict__
{'_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()
result
[(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()
result
[('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:
session.add(gt)
session.commit()
Connections
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()
conn.execute(...)
trans = conn.begin()
conn.execute('INSERT INTO MY_TABLE ...)
trans.commit()
with engine.connect() as con:
response = con.execute(statement)
with engine.connect() as con:
response = con.execute("select * from customers")
response
<sqlalchemy.engine.cursor.LegacyCursorResult at 0x20068fa2e50>
result = response.fetchall()
result
[('John Smith', 'js@gmail.com', 100, 25),
('Karen Smith', 'ks@gmail.com', 101, 42),
('John Smith', 'js@gmail.com', 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:
sess.merge(rec)
try:
sess.commit()
except Exception as ex:
print(ex)
sess.rollback()
Or you could pass one sess to the whole thing.
Synactic Sugar and Tricks
Get the date from a timestamp
date(my_timestamp)
Optimizing
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