Skip to content

Instantly share code, notes, and snippets.

@dckc
Created November 8, 2011 22:13
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dckc/1349445 to your computer and use it in GitHub Desktop.
Save dckc/1349445 to your computer and use it in GitHub Desktop.
SQLAlchemy test case that worked in 0.7.2 and doesn't in 0.7.3
from sqlalchemy import Table, Column, text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.types import INTEGER, VARCHAR, TEXT
from sqlalchemy.sql import select, and_, func
Base = declarative_base()
institutions = ('kuh', 'kupi', 'kumc')
redcap_data = Table('redcap_data', Base.metadata,
Column(u'project_id', INTEGER(),
nullable=False, default=text(u"'0'"),
primary_key=True),
Column(u'event_id', INTEGER(), primary_key=True),
Column(u'record', VARCHAR(length=100), primary_key=True),
Column(u'field_name', VARCHAR(length=100),
primary_key=True),
Column(u'value', TEXT()),
)
def _sponsor_queries(oversight_project_id):
'''
>>> from pprint import pprint
>>> decision, candidate, cdwho = _sponsor_queries(123)
>>> print str(decision)
... # doctest: +NORMALIZE_WHITESPACE
SELECT redcap_data.project_id, redcap_data.record,
redcap_data.value AS decision, count(*) AS count_1 FROM
redcap_data WHERE redcap_data.field_name LIKE :field_name_1 AND
redcap_data.project_id = :project_id_1 GROUP BY
redcap_data.project_id, redcap_data.record, redcap_data.value
HAVING count(*) = :count_2
>>> pprint(decision.compile().params)
{u'count_2': 3, u'field_name_1': 'approve_%', u'project_id_1': 123}
>>> print str(candidate)
... # doctest: +NORMALIZE_WHITESPACE
SELECT redcap_data.project_id, redcap_data.record,
redcap_data.value AS userid FROM redcap_data WHERE
redcap_data.field_name LIKE :field_name_1
>>> pprint(candidate.compile().params)
{u'field_name_1': 'user_id_%'}
This works on 0.7.2:
>>> print str(cdwho) # doctest: +NORMALIZE_WHITESPACE
SELECT cd_record AS record,
cd_decision AS decision,
who_userid AS candidate,
expire_dt_exp AS dt_exp
FROM
(SELECT cd.project_id AS cd_project_id,
cd.record AS cd_record,
cd.decision AS cd_decision,
cd.count_1 AS cd_count_1,
who.project_id AS who_project_id,
who.record AS who_record,
who.userid AS who_userid,
expire.project_id AS expire_project_id,
expire.record AS expire_record,
expire.dt_exp AS expire_dt_exp
FROM
(SELECT redcap_data.project_id AS project_id,
redcap_data.record AS record,
redcap_data.value AS decision, count(*) AS count_1
FROM redcap_data
WHERE redcap_data.field_name LIKE :field_name_1
AND redcap_data.project_id = :project_id_1
GROUP BY redcap_data.project_id, redcap_data.record,
redcap_data.value
HAVING count(*) = :count_2) AS cd
JOIN
(SELECT redcap_data.project_id AS project_id,
redcap_data.record AS record,
redcap_data.value AS userid
FROM redcap_data
WHERE redcap_data.field_name LIKE :field_name_2) AS who
ON who.record = cd.record
AND who.project_id = cd.project_id
LEFT OUTER JOIN (SELECT redcap_data.project_id AS project_id,
redcap_data.record AS record,
redcap_data.value AS dt_exp
FROM redcap_data
WHERE redcap_data.field_name = :field_name_3)
AS expire
ON expire.record = cd.record AND expire.project_id = cd.project_id)
AS cdwho
>>> pprint(cdwho.compile().params)
{u'count_2': 3,
u'field_name_1': 'approve_%',
u'field_name_2': 'user_id_%',
u'field_name_3': 'date_of_expiration',
u'project_id_1': 123}
'''
rd = redcap_data
# committee decisions
decision = select((rd.c.project_id, rd.c.record,
rd.c.value.label('decision'),
func.count())).where(
and_(rd.c.field_name.like('approve_%'),
rd.c.project_id==oversight_project_id)
).group_by(rd.c.project_id,
rd.c.record,
rd.c.value).having(
func.count() == len(institutions)).alias('cd')
# todo: consider combining record, event, project_id into one attr
candidate = select((rd.c.project_id, rd.c.record,
rd.c.value.label('userid'))).where(
rd.c.field_name.like('user_id_%')).alias('who')
dt_exp = select((rd.c.project_id, rd.c.record,
rd.c.value.label('dt_exp'))).where(
rd.c.field_name == 'date_of_expiration').alias('expire')
j = decision.join(candidate,
and_(candidate.c.record == decision.c.record,
candidate.c.project_id == decision.c.project_id)
).outerjoin(dt_exp, and_(
dt_exp.c.record == decision.c.record,
dt_exp.c.project_id == decision.c.project_id)
).alias('cdwho').select()
cdwho = j.with_only_columns((j.c.cd_record.label('record'),
j.c.cd_decision.label('decision'),
j.c.who_userid.label('candidate'),
j.c.expire_dt_exp.label('dt_exp')))
return decision, candidate, cdwho
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment