pg_stud.utils

Module Contents

Functions

set_search_path	Sets cursor’s search_path to the schema of the topic
is_installed	Checks if the schema of the topic is install on the user database Args: conn: Connection with autocommit off
install_db	Installs the topic’s linked script by executing it in a schema named after the topic
uninstall_db	Uninstalls the topic schema by droping it.
execute	Executes a query on the connection.
execute_check	Executes queries with only produce a boolean If an Exception occurs False is returned.
update_user_exercise	Updates the specified UserExercise with buffer and optionally is_correct.
do_reset_db	Resets db to specific exercise
dict_row	Row factory to represent rows as dictionaries. Copied from psycopg3 With support for columns with the same name.
numerate_duplicates
flatten_list	Flattens a list.

API

pg_stud.utils.set_search_path(cursor: psycopg.cursor.Cursor, topic: exercises.models.Topic) → None

Sets cursor’s search_path to the schema of the topic

pg_stud.utils.is_installed(conn: psycopg.connection.Connection, topic: exercises.models.Topic) → bool

Checks if the schema of the topic is install on the user database Args: conn: Connection with autocommit off

pg_stud.utils.install_db(cursor: psycopg.cursor.Cursor, topic: exercises.models.Topic) → None

Installs the topic’s linked script by executing it in a schema named after the topic

pg_stud.utils.uninstall_db(cursor: psycopg.cursor.Cursor, topic: exercises.models.Topic) → None

Uninstalls the topic schema by droping it.

pg_stud.utils.execute(conn: psycopg.connection.Connection, query: str, topic: exercises.models.Topic) → List[Dict[str, Any]]

Executes a query on the connection.

Returns: If the query has an error [{‘error_in_query’: e.args}] is returned. For non-SELECT-like queries which do not produce output [{‘no_output’: e.args}] is returned. When no row is returned because there was no entry found a list with a dict column to “” is returned >>> execute(conn, “SELECT * FROM table”) [{“column1”: “1”, “column2”: “2”}, {“column1”: “3”, “column2”: None}] >>> execute(conn, “BAD SQL STATEMENT”) [{“error_in_query”: “Syntaxfehler bei »ERROR« LINE 1: ERROR ^”}] >>> execute(conn, “SELECT * FROM table where column1=’not_exists’”) [{“column1”: “”, “column2”: “”}]

pg_stud.utils.execute_check(conn: psycopg.connection.Connection, query: str, topic: exercises.models.Topic) → bool

Executes queries with only produce a boolean If an Exception occurs False is returned.

pg_stud.utils.update_user_exercise(user: ltiapi.models.LTIUser, exercise: exercises.models.Exercise, buffer_save: str, is_correct: Optional[bool] = None)

Updates the specified UserExercise with buffer and optionally is_correct.

pg_stud.utils.do_reset_db(conn: psycopg.connection.Connection, topic: exercises.models.Topic, exercise: exercises.models.Exercise)

Resets db to specific exercise

Args: conn (Connection): Connection, needs to be fresh topic (m.Topic): topic exercise (m.Exercise): exercise

pg_stud.utils.dict_row(cursor: psycopg.cursor.BaseCursor[Any, Any]) → psycopg.rows.RowMaker[psycopg.rows.DictRow]

Row factory to represent rows as dictionaries. Copied from psycopg3 With support for columns with the same name.

The dictionary keys are taken from the column names of the returned columns.

pg_stud.utils.numerate_duplicates(input_list: Iterable[str])

pg_stud.utils.flatten_list(input_list: Iterable[List[Any] | Any])

Flattens a list.