Created
November 8, 2011 22:13
-
-
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
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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