Module `dbops.selects`

db select functions jobs

Functions

def batch_infos_for_job_dict(extracted_dict: dict[str, dict[str, typing.Any]], provider_facility: str) ‑> list[BatchInfo]: given a dict of case dicts, return the id, dos, and state for batches that match with the supplied cases
def existing_entities_with_new_as_pdf(pt_dict: dict[str, typing.Any]) ‑> dict[str, str | list[dict[str, str | bool]]]: get the entities currently attached to the case and append a newly received doc in the form of a PDF. Used when new docs are received after the case has been charged entered. pt_dict must contain a valid extracted documentation set (at least one json type doc) and a valid job id or an error will result.
def job_library(provider_facility: str, dates_of_service: collections.abc.Sequence[str], job_ids: collections.abc.Sequence[int] = (-1,)) ‑> dict[str, dict[str, typing.Any]]: get a dictionary of jobs for a batch specified by facility and date of service. Principle use case is for matching previously created jobs that have no documents to newly extracted documents as they come in.
def simple_select(select_list: str, table_expression: str, sort_specification: str = '', *, params: tuple[typing.Any, ...] | dict[str, typing.Any] = (), formatter: collections.abc.Callable[[collections.abc.Sequence[tuple[typing.Any, ...]]], ~T] = <function _formatter_default>) ‑> ~T: execute a generic query and return the results. use the formatter kwarg to specify a custom output type, e.g. if cursor.fetchall() returns a list of 2 length tuples with unique hashables in the first position, passing formatter=dict will return a dict with appropriate typing. The first 3 arguments will be complied into the select query as follows: f"SELECT {select_list} FROM {table_expression} {sort_specification}" Refer to https://www.postgresql.org/docs/10/queries-overview.html. Note that 'sort_expression' includes everything after the table_expression including all where conditions, group bys, and order bys. The caller MUST supply all required keywords (i.e. "WHERE", "GROUP BY", etc.).
def simple_select_from_common(select_list: str, table_expression: str, sort_specification: str = '', *, params: tuple[typing.Any, ...] | dict[str, typing.Any] = (), formatter: collections.abc.Callable[[collections.abc.Sequence[tuple[typing.Any, ...]]], ~T] = <function _formatter_default>) ‑> ~T: Same as simple_select() defined above but targets common_db in place of facility specific db. see simple_select() docstring for more info.
def supersede_existing_with_new_entities(pt_dict: dict[str, typing.Any]): Creates a SUPERSEDED ON [date] combined doc from all existing json (extracted) entities and appends this and all previously attached pdfs to the end of the array of new extracted docs. Modifies pt_dict in place.
def unmatched_pdf_reference(provider_facility: str, dates_of_service: collections.abc.Sequence[str], document_id_not_like: str = '%_combined.pdf', state_in: collections.abc.Sequence[str] = ('New', 'InProgress', 'OnHold')) ‑> dict[str, utilities.library_utils.PDFLibReference]: get all PDFs that have NOT been combined (i.e. documentId does not end in '_combined.pdf'). Allows intraop records and facesheets to be combined and re-extracted when both documents don't post on the same day. nphllc.com's chaph facility is the principal use case.