"""Classlist database access.
Code to run a query to generate a classlist data file.
"""
from uw.dbtools import copy_to_csv
from uw.sql.query import query_and, query_or
[docs]class BadColumnsError (ValueError):
def __init__ (self, allowed_columns, bad_columns):
super ().__init__ ('Unknown columns: ' + ', '.join (bad_columns) +
'\n' + 'Allowed columns: ' + ', '.join (allowed_columns))
self.allowed_columns = allowed_columns
self.bad_columns = bad_columns
fixed_columns = [
'uw_id', 'userid', 'surname', 'givennames',
'group_code', 'program_code', 'plan_code',
]
[docs]def classlist_csv (cursor, term, admin_id, dropped=False, header=True, columns=None, userids=None, uw_ids=None):
values = {
'term_id': term.numericCode (),
'admin_id': admin_id,
'dropped': "not null" if dropped else "null",
}
divisions = cursor.execute_values ("select quote_ident (division_code) from division_division where (term_id, admin_id) = (%(term_id)d, %(admin_id)d) order by division_code" % values)
default_columns = fixed_columns + divisions
if columns is None:
columns = default_columns
else:
allowed_columns = set (default_columns)
bad_columns = [c for c in columns if not c in allowed_columns]
if bad_columns:
raise BadColumnsError (default_columns, bad_columns)
if userids is None:
userid_query = None
else:
userids = cursor.execute_values ("select quote_literal (t) from unnest (%(userids)s) as t", userids=userids)
userid_query = f"userid in ({', '.join (userids)})"
if uw_ids is None:
uw_id_query = None
else:
uw_ids = cursor.execute_values ("select quote_literal (t) from unnest (%(uw_ids)s) as t", uw_ids=uw_ids)
uw_id_query = f"uw_id in ({', '.join (uw_ids)})"
values.update ({
'where_clause': query_and (
"(term_id, admin_id) = (%(term_id)d, %(admin_id)d)",
"dropped is %(dropped)s",
query_or (userid_query, uw_id_query)
) % values,
'columns': ', '.join (columns),
'div_types': ''.join (", %s text" % d for d in divisions),
'base_query': "select person_id, division_seq, division_value from division_student where (term_id, admin_id) = (%(term_id)d, %(admin_id)d) order by 1" % values,
'division_query': "select division_seq from division_division where (term_id, admin_id) = (%(term_id)d, %(admin_id)d) order by division_code" % values,
})
if divisions:
values['crosstab'] = " left join crosstab ($$%(base_query)s$$, $$%(division_query)s$$) as ct (person_id integer%(div_types)s) using (person_id)" % values
else:
values['crosstab'] = ""
classlist_query = "select %(columns)s from registration_classlist%(crosstab)s where %(where_clause)s order by surname, givennames, uw_id" % values
result = copy_to_csv (cursor, classlist_query, header=header)
return result.getvalue ()