Skip to content

Instantly share code, notes, and snippets.

@tuantranf
Forked from sirex/main.py
Created April 29, 2020 03:12
Show Gist options
  • Save tuantranf/0ad3c504beefbf646d49c742c055a844 to your computer and use it in GitHub Desktop.
Save tuantranf/0ad3c504beefbf646d49c742c055a844 to your computer and use it in GitHub Desktop.
sqlalchemy joins with AS
import sqlalchemy as sa
import sqlparse
metadata = sa.MetaData()
tasks = sa.Table(
'tasks', metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('bot', sa.String(255), nullable=False),
sa.Column('task', sa.String(255), nullable=False),
)
state = sa.Table(
'state', metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('source_id', sa.Integer, sa.ForeignKey(tasks.c.id)),
sa.Column('target_id', sa.Integer, sa.ForeignKey(tasks.c.id), nullable=False),
)
errors = sa.Table(
'errors', metadata,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('state_id', sa.Integer, sa.ForeignKey(state.c.id), nullable=False),
sa.Column('row_id', sa.Integer, nullable=False),
)
source = state.alias().join(tasks.alias('source'), state.c.source_id == tasks.alias('source').c.id)
target = state.alias().join(tasks.alias('target'), state.c.target_id == tasks.alias('target').c.id)
query = (
sa.select([
errors.c.id,
source.c.source_task.label('source'),
target.c.target_task.label('target'),
]).
select_from(errors.join(state, errors.c.state_id == state.c.id)).
select_from(source).
select_from(target)
)
print(sqlparse.format(str(query), reindent=True))
# OUTPUT:
#
# SELECT errors.id,
# source.task AS source,
# target.task AS target
# FROM errors
# JOIN state ON errors.state_id = state.id, state AS state_1
# JOIN tasks AS source ON state.source_id = source.id, state AS state_2
# JOIN tasks AS target ON state.target_id = target.id
import sqlparse
import sqlalchemy as sa
meta = sa.MetaData()
a = sa.Table(
'a', meta,
sa.Column('id', sa.Integer, primary_key=True),
)
b = sa.Table(
'b', meta,
sa.Column('id', sa.Integer, primary_key=True),
sa.Column('x', sa.Integer, sa.ForeignKey(a.c.id)),
sa.Column('y', sa.Integer, sa.ForeignKey(a.c.id)),
)
x = b.alias('x')
y = b.alias('y')
query = (
sa.select(['*']).
select_from(a.join(x, a.c.id == x.c.x)).
select_from(a.join(y, a.c.id == y.c.y))
)
print(sqlparse.format(str(query), reindent=True))
# OUTPUT:
#
# SELECT *
# FROM a
# JOIN b AS x ON a.id = x.x,
# a
# JOIN b AS y ON a.id = y.y
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment