15.2.116. camcops_server.cc_modules.cc_exportmodels

camcops_server/cc_modules/cc_exportmodels.py

Copyright (C) 2012, University of Cambridge, Department of Psychiatry. Created by Rudolf Cardinal (rnc1001@cam.ac.uk).

This file is part of CamCOPS.

CamCOPS is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version.

CamCOPS is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with CamCOPS. If not, see <https://www.gnu.org/licenses/>.

Define models for export functions (e.g. HL7, file-based export).

class camcops_server.cc_modules.cc_exportmodels.ExportedTask(recipient=None, task=None, basetable=None, task_server_pk=None, *args, **kwargs)[source]

Class representing an attempt to exported a task (as part of a ExportRun) to a specific camcops_server.cc_modules.cc_exportrecipient.ExportRecipient.

__init__(recipient=None, task=None, basetable=None, task_server_pk=None, *args, **kwargs)

Can initialize with a task, or a basetable/task_server_pk combination.

Parameters

(However, we must also support a no-parameter constructor, not least for our merge_db() function.)

abort(msg: str) → None[source]

Record failure, and why.

(Called abort not fail because PyCharm has a bug relating to functions named fail: https://stackoverflow.com/questions/21954959/pycharm-unreachable-code.)

Parameters

msg – why

cancel() → None[source]

Marks the task export as cancelled/invalidated.

May trigger a resend (which is the point).

export(req: CamcopsRequest) → None[source]

Performs an export of the specific task.

Parameters

req – a camcops_server.cc_modules.cc_request.CamcopsRequest

export_file(filename: str, text: Optional[str] = None, binary: Optional[bytes] = None, text_encoding: str = 'utf8') → bool[source]

Exports a file.

Parameters

  • filename –

  • text – text contents (specify this XOR binary)

  • binary – binary contents (specify this XOR text)

  • text_encoding – encoding to use when writing text

Returns: was it exported?

property filegroup: camcops_server.cc_modules.cc_exportmodels.ExportedTaskFileGroup

Returns a ExportedTaskFileGroup, creating it if necessary.

finish() → None[source]

Records the finish time.

init_on_load() → None[source]

Called when SQLAlchemy recreates an object; see https://docs.sqlalchemy.org/en/latest/orm/constructors.html.

succeed() → None[source]

Register success.

property task: Task

Returns the associated task.

classmethod task_already_exported(dbsession: sqlalchemy.orm.session.Session, recipient_name: str, basetable: str, task_pk: int) → bool[source]

Has the specified task already been successfully exported?

Parameters

  • dbsession – a sqlalchemy.orm.session.Session

  • recipient_name –

  • basetable – name of the task’s base table

  • task_pk – server PK of the task

Returns

does a successful export record exist for this task?

class camcops_server.cc_modules.cc_exportmodels.ExportedTaskEmail(exported_task=None)[source]

Represents an individual email export.

__init__(exported_task=None)
Parameters

exported_task – ExportedTask object

export_task(req: CamcopsRequest) → None[source]

Exports the task itself to an email.

Parameters

req – a camcops_server.cc_modules.cc_request.CamcopsRequest

class camcops_server.cc_modules.cc_exportmodels.ExportedTaskFhir(exported_task=None)[source]

Represents an individual FHIR export.

__init__(exported_task=None)
Parameters

exported_task – ExportedTask object

export_task(req: CamcopsRequest) → None[source]

Exports the task to FHIR.

Parameters

req – a camcops_server.cc_modules.cc_request.CamcopsRequest

class camcops_server.cc_modules.cc_exportmodels.ExportedTaskFhirEntry(**kwargs)[source]

Details of Patients, Questionnaires, QuestionnaireResponses exported to a FHIR server for a single task.

__init__(**kwargs)

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

property location_url: str

Puts the FHIR server API URL together with the returned location, so we can hyperlink to the resource.

class camcops_server.cc_modules.cc_exportmodels.ExportedTaskFileGroup(exported_task=None)[source]

Represents a small set of files exported in relation to a single task.

__init__(exported_task=None)
Parameters

exported_task – ExportedTask object

abort(msg: str) → None[source]

Record failure, and why.

(Called abort not fail because PyCharm has a bug relating to functions named fail: https://stackoverflow.com/questions/21954959/pycharm-unreachable-code.)

Parameters

msg – why

export_file(filename: str, text: Optional[str] = None, binary: Optional[bytes] = None, text_encoding: str = 'utf8') → False[source]

Exports the file.

Parameters

  • filename –

  • text – text contents (specify this XOR binary)

  • binary – binary contents (specify this XOR text)

  • text_encoding – encoding to use when writing text

Returns

was it exported?

Return type

bool

export_task(req: CamcopsRequest) → None[source]

Exports the task itself to a file.

Parameters

req – a camcops_server.cc_modules.cc_request.CamcopsRequest

finish_run_script_if_necessary() → None[source]

Completes the file export by running the external script, if required.

note_exported_file(*filenames: str) → None[source]

Records a filename that has been exported, or several.

Parameters

*filenames – filenames

succeed() → None[source]

Register success.

class camcops_server.cc_modules.cc_exportmodels.ExportedTaskHL7Message(exported_task=None, *args, **kwargs)[source]

Represents an individual HL7 message.

__init__(exported_task=None, *args, **kwargs)

Must support parameter-free construction, not least for merge_db().

abort(msg: str, diverted_not_sent: bool = False) → None[source]

Record that we failed, and so did our associated task export.

(Called abort not fail because PyCharm has a bug relating to functions named fail: https://stackoverflow.com/questions/21954959/pycharm-unreachable-code.)

Parameters

  • msg – reason for failure

  • diverted_not_sent – deliberately diverted (and not counted as sent) rather than a sending failure?

divert_to_file(req: CamcopsRequest) → None[source]

Write an HL7 message to a file. For debugging.

Parameters

req – a camcops_server.cc_modules.cc_request.CamcopsRequest

export_task(req: CamcopsRequest) → None[source]

Exports the task itself to an HL7 message.

Parameters

req – a camcops_server.cc_modules.cc_request.CamcopsRequest

init_on_load() → None[source]

Called when SQLAlchemy recreates an object; see https://docs.sqlalchemy.org/en/latest/orm/constructors.html.

make_hl7_message(req: CamcopsRequest) → None[source]

Makes an HL7 message and stores it in self._hl7_msg.

May also store it in self.message (which is saved to the database), if we’re saving HL7 messages.

See

static ping_hl7_server(recipient: camcops_server.cc_modules.cc_exportrecipient.ExportRecipient) → bool[source]

Performs a TCP/IP ping on our HL7 server; returns success. If we’ve already pinged successfully during this run, don’t bother doing it again.

(No HL7 PING method yet. Proposal is http://hl7tsc.org/wiki/index.php?title=FTSD-ConCalls-20081028 So use TCP/IP ping.)

Parameters

recipient – an camcops_server.cc_modules.cc_exportrecipient.ExportRecipient

Returns

success

Return type

bool

succeed(now: Optional[pendulum.datetime.DateTime] = None) → None[source]

Record that we succeeded, and so did our associated task export.

static task_acceptable_for_hl7(recipient: camcops_server.cc_modules.cc_exportrecipient.ExportRecipient, task: Task) → bool[source]

Is the task valid for HL7 export. (For example, anonymous tasks and tasks missing key ID information may not be.)

Parameters
Returns

valid?

Return type

bool

transmit_hl7() → None[source]

Sends the HL7 message over TCP/IP.

valid() → bool[source]

Checks for internal validity; returns a bool.

class camcops_server.cc_modules.cc_exportmodels.ExportedTaskRedcap(exported_task=None)[source]

Represents an individual REDCap export.

__init__(exported_task=None)
Parameters

exported_task – ExportedTask object

export_task(req: CamcopsRequest) → None[source]

Exports the task to REDCap.

Parameters

req – a camcops_server.cc_modules.cc_request.CamcopsRequest

camcops_server.cc_modules.cc_exportmodels.gen_exportedtasks(collection: camcops_server.cc_modules.cc_taskcollection.TaskCollection) → Generator[camcops_server.cc_modules.cc_exportmodels.ExportedTask, None, None][source]

Generates task export entries from a collection.

Parameters

collection – a camcops_server.cc_modules.cc_taskcollection.TaskCollection

Yields

ExportedTask objects

camcops_server.cc_modules.cc_exportmodels.gen_tasks_having_exportedtasks(collection: camcops_server.cc_modules.cc_taskcollection.TaskCollection) → Generator[Task, None, None][source]

Generates tasks from a collection, creating export logs as we go.

Used for database exports.

Parameters

collection – a camcops_server.cc_modules.cc_taskcollection.TaskCollection

Yields

camcops_server.cc_modules.cc_task.Task objects

camcops_server.cc_modules.cc_exportmodels.get_collection_for_export(req: CamcopsRequest, recipient: camcops_server.cc_modules.cc_exportrecipient.ExportRecipient, via_index: bool = True, debug: bool = False) → camcops_server.cc_modules.cc_taskcollection.TaskCollection[source]

Returns an appropriate task collection for this export recipient, namely those tasks that are desired and (in the case of incremental exports) haven’t already been sent.

“Not already sent” means “not already sent to an export recipient with the same name (even if other aspects of the export recipient have changed)”.

Parameters
Returns

a camcops_server.cc_modules.cc_taskcollection.TaskCollection