Mosaiq

Mosaiq#

Summary#

Module to handle the connection to a Mosaiq SQL database and execute SQL queries.

API#

Connect and Query the Mosaiq Database#

pymedphys.mosaiq.connect(hostname: str, port: int = 1433, database: str = 'MOSAIQ', alias: str | None = None, username: str | None = None, password: str | None = None) → Connection[source]#

Connect to a Mosaiq SQL server.

When neither username nor password is supplied, credentials are read from the operating system’s credential storage, with an interactive prompt for missing values. Credentials entered at that prompt are stored for later use. Alternatively, supply both username and password explicitly; these are used directly without storing them. Supplying only one raises an error.

Can optionally be called as a context manager. This will close the database connection once leaving the context manager.

Parameters:

  • hostname (str) – The IP address or hostname of the SQL server.

  • port (int, optional) – The port at which the SQL server is hosted, by default 1433

  • database (str, optional) – The MSSQL database name, by default "MOSAIQ"

  • alias (Optional[str], optional) – A human readable representation of the server, this is the name of the server presented to the user should there not be credentials already on the machine, by default "hostname:port/database"

  • username (Optional[str], optional) – Provide a username to login to the database with, by default the username is either pulled from the system’s credential storage, or an interactive prompt is used.

  • password (Optional[str], optional) – Provide a password to login to the database with, by default the password is either pulled from the system’s credential storage, or an interactive prompt is used.

Returns:

connection – A database connection. This connection can be passed to pymedphys.mosaiq.execute() to be able to run queries.

The method close() can be called on this object to close the database connection.

Return type:

pymedphys.mosaiq.Connection

Examples

See pymedphys.mosaiq.execute() for examples of usage.

pymedphys.mosaiq.execute(connection: Connection, query: str, parameters: dict | None = None) → list[tuple[str, ...]][source]#

Execute SQL queries on a Mosaiq database.

Parameters:

  • connection (pymedphys.mosaiq.Connection) – A database connection. This can be retrieved by calling pymedphys.mosaiq.connect

  • query (str) – The SQL query to execute. Use pymssql placeholders such as %(last_name)s and pass the values separately in parameters, for example {"last_name": "PHANTOM"}. Do not interpolate values with f-strings, str.format, or Python’s percent operator. The underlying driver handles parameter substitution.

  • parameters (Dict, optional) – Parameters to be included within the query. These are sanitised by the underlying pymssql library before being included within the query string, by default None

Returns:

results – The results from the database query organised so that each row is an item within the returned list.

Return type:

List[Tuple[str, …]]

Examples

Directly calling the connection object and listing all patients that have the last name of "PHANTOM".

>>> import pymedphys.mosaiq
>>> connection = pymedphys.mosaiq.connect('msqsql')  

>>> pymedphys.mosaiq.execute(
...     connection,
...     '''
...     SELECT
...         Ident.IDA,
...         Patient.Last_Name,
...         Patient.First_Name
...     FROM Ident, Patient
...     WHERE
...         Patient.Pat_ID1 = Ident.Pat_ID1 AND
...         Patient.Last_Name = %(last_name)s
...     ''',
...     {"last_name": "PHANTOM"},
... )  
[('654324', 'PHANTOM', 'CATPHAN'),
 ('944444', 'PHANTOM', 'DELTA4'),
 ('654321', 'PHANTOM', 'PELVIS'),
 ('012534', 'PHANTOM', 'QUASAR4D'),
 ('987654', 'PHANTOM', 'RESPIRATORY'),
 ('654325', 'PHANTOM', 'RW3')]

Connecting via a context manager and using pandas to present the results as a table.

>>> import pandas as pd
>>> import pymedphys.mosaiq

>>> with pymedphys.mosaiq.connect('msqsql') as connection:  
...     results = pymedphys.mosaiq.execute(
...         connection,
...         '''
...         SELECT
...             Ident.IDA,
...             Chklist.Due_DtTm
...         FROM Chklist, Staff, Ident
...         WHERE
...             Chklist.Pat_ID1 = Ident.Pat_ID1 AND
...             Staff.Staff_ID = Chklist.Rsp_Staff_ID AND
...             Staff.Last_Name = %(qcl_location)s AND
...             Chklist.Complete = 0
...         ''',
...         {"qcl_location": "Physics_Check"},
...     )

>>> pd.DataFrame(  
...     data=results,
...     columns=[
...         "patient_id",
...         "due",
...     ],
... )
  patient_id                 due
0     000000 2021-02-01 23:59:59
1     000001 2021-02-01 23:59:59
2     000002 2021-02-01 23:59:59
3     000003 2021-03-08 23:59:59