`nexusLIMS.db.models`#

SQLModel database models for NexusLIMS.

This module defines the SQLModel ORM classes that map to the NexusLIMS database tables (instruments and session_log).

Module Contents#

Classes#

`TZDateTime`	Custom DateTime type that preserves timezone information in SQLite.
`Instrument`	Instrument configuration from the NexusLIMS database.
`SessionLog`	Individual session log entry (START, END, or RECORD_GENERATION event).
`UploadLog`	Log of export attempts to destination repositories.
`ExternalUserIdentifier`	Maps NexusLIMS usernames to external system user IDs.

Functions#

`get_external_id`	Get external system ID for a NexusLIMS user.
`get_nexuslims_username`	Reverse lookup: find NexusLIMS username from external ID.
`store_external_id`	Store or update external ID mapping.
`get_all_external_ids`	Get all external IDs for a user.

API#

class nexusLIMS.db.models.TZDateTime(*args: Any, **kwargs: Any)[source]#

Bases: sqlalchemy.types.TypeDecorator

Custom DateTime type that preserves timezone information in SQLite.

SQLite stores datetimes as TEXT and doesn’t preserve timezone info. This TypeDecorator stores timezone-aware datetimes as ISO-8601 strings with timezone offset, and restores them as timezone-aware datetime objects.

impl#

cache_ok#: True

process_bind_param(value, dialect)[source]#: Convert timezone-aware datetime to ISO string for storage.

process_result_value(value, dialect)[source]#: Convert ISO string back to timezone-aware datetime.

class nexusLIMS.db.models.Instrument(**data: Any)[source]#

Bases: sqlmodel.SQLModel

Instrument configuration from the NexusLIMS database.

Represents an electron microscopy instrument in the facility, with configuration for calendar integration, file storage, and metadata.

Parameters:

instrument_pid – Unique identifier for the instrument (e.g., “FEI-Titan-TEM-012345”)
api_url – Calendar API endpoint URL for this instrument’s scheduler (e.g., https://<nemo_address>/api/tools/?id=<tool_id>)
calendar_url – URL to the instrument’s web-accessible calendar
location – Physical location (building and room number)
display_name – Human-readable instrument name displayed in NexusLIMS records
property_tag – Unique numeric identifier (for reference)
filestore_path – Relative path under NX_INSTRUMENT_DATA_PATH where data is stored
harvester – Harvester module to use (“nemo” or “sharepoint”)
timezone_str – IANA timezone database string (e.g., “America/New_York”)

instrument_pid#: Type: str

api_url#: Type: str

calendar_url#: Type: str

location#: Type: str

display_name#: Type: str

property_tag#: Type: str

filestore_path#: Type: str

harvester#: Type: str

timezone_str#: Type: str

session_logs#

‘Relationship(…)’

Type: list[SessionLog]

property name: str#: Alias for instrument_pid (backward compatibility).

property timezone: BaseTzInfo#: Convert timezone string to pytz timezone object.

localize_datetime(_dt: datetime) → datetime[source]#

Localize a datetime to an Instrument’s timezone.

Convert a date and time to the timezone of this instrument. If the supplied datetime is naive (i.e. does not have a timezone), it will be assumed to already be in the timezone of the instrument, and the displayed time will not change. If the timezone of the supplied datetime is different than the instrument’s, the time will be adjusted to compensate for the timezone offset.

Parameters:: _dt – The datetime object to localize
Returns:: A datetime object with the same timezone as the instrument
Return type:: datetime

localize_datetime_str(_dt: datetime, fmt: str = '%Y-%m-%d %H:%M:%S %Z') → str[source]#

Localize a datetime to an Instrument’s timezone and return as string.

Convert a date and time to the timezone of this instrument, returning a textual representation of the object, rather than the datetime itself. Uses localize_datetime() for the actual conversion.

Parameters:

_dt – The datetime object to localize
fmt – The strftime format string to use to format the output

Returns:

The formatted textual representation of the localized datetime

Return type:

str

to_dict() → dict[source]#

Return a dictionary representation of the Instrument object.

Handles special cases like renaming ‘instrument_pid’ and converting timezone objects to strings.

Returns:: A dictionary representation of the instrument, suitable for database insertion or JSON serialization.
Return type:: dict

to_json(**kwargs) → str[source]#

Return a JSON string representation of the Instrument object.

Parameters:: **kwargs – Additional keyword arguments to pass to json.dumps.
Returns:: A JSON string representation of the instrument.
Return type:: str

class nexusLIMS.db.models.SessionLog(**data: Any)[source]#

Bases: sqlmodel.SQLModel

Individual session log entry (START, END, or RECORD_GENERATION event).

A simple mapping of one row in the session_log table. Each session typically has a START and END log with matching session_identifier, and may have additional RECORD_GENERATION logs.

Parameters:

session_identifier – A unique string consistent among a single record’s START, END, and RECORD_GENERATION events (often a UUID)
instrument – The instrument associated with this session (foreign key reference to instruments table)
timestamp – The datetime representing when the event occurred
event_type – The type of log (START, END, or RECORD_GENERATION)
user – The username associated with this session (if known)
record_status – The status for this record (defaults to WAITING_FOR_END)

id_session_log#: Type: int | None

session_identifier#: Type: str

instrument#: Type: str

timestamp#: Type: datetime.datetime

event_type#: Type: nexusLIMS.db.enums.EventType

record_status#: Type: nexusLIMS.db.enums.RecordStatus

user#: Type: str | None

instrument_obj#

‘Relationship(…)’

Type: nexusLIMS.db.models.Instrument | None

insert_log() → bool[source]#

Insert this log into the NexusLIMS database.

Inserts a log into the database with the information contained within this SessionLog’s attributes (used primarily for NEMO usage_event integration). It will check for the presence of a matching record first and warn without inserting anything if it finds one.

Returns:: success – Whether or not the session log row was inserted successfully
Return type:: bool

class nexusLIMS.db.models.UploadLog(**data: Any)[source]#

Bases: sqlmodel.SQLModel

Log of export attempts to destination repositories.

Tracks per-destination export results for each session, enabling multi-destination export with granular success/failure tracking.

Parameters:

id – Auto-incrementing primary key
session_identifier – Foreign key reference to session_log.session_identifier
destination_name – Name of the export destination (e.g., “cdcs”, “labarchives”)
success – Whether the export succeeded
record_id – Destination-specific record identifier (if successful)
record_url – Direct URL to view the exported record (if successful)
error_message – Error message if export failed
timestamp – When the export attempt occurred
metadata_json – JSON-serialized metadata dict with destination-specific details

id#: Type: int | None

session_identifier#: Type: str

destination_name#: Type: str

success#: Type: bool

timestamp#: Type: datetime.datetime

record_id#: Type: str | None

record_url#: Type: str | None

error_message#: Type: str | None

metadata_json#: Type: str | None

class nexusLIMS.db.models.ExternalUserIdentifier(**data: Any)[source]#

Bases: sqlmodel.SQLModel

Maps NexusLIMS usernames to external system user IDs.

Maintains a star topology with nexuslims_username (from session_log.user) as the canonical identifier, mapping to external system IDs.

Parameters:

id – Auto-incrementing primary key
nexuslims_username – Canonical username in NexusLIMS (from session_log.user)
external_system – External system identifier (nemo, labarchives_eln, etc.)
external_id – User ID/username in the external system
email – User’s email for verification/matching (optional)
created_at – When this mapping was created
last_verified_at – Last time this mapping was verified (optional)
notes – Additional notes about this mapping (optional)

Examples:

>>> # NEMO harvester user ID
>>> ExternalUserIdentifier(
...     nexuslims_username='jsmith',
...     external_system=ExternalSystem.NEMO,
...     external_id='12345'
... )

>>> # LabArchives UID from OAuth
>>> ExternalUserIdentifier(
...     nexuslims_username='jsmith',
...     external_system=ExternalSystem.LABARCHIVES_ELN,
...     external_id='285489257Ho...',
...     email='jsmith@upenn.edu'
... )

id#: Type: int | None

nexuslims_username#: Type: str

external_system#: Type: str

external_id#: Type: str

email#: Type: str | None

created_at#: Type: datetime.datetime

last_verified_at#: Type: datetime.datetime | None

notes#: Type: str | None

nexusLIMS.db.models.get_external_id(nexuslims_username: str, external_system: ExternalSystem) → str | None[source]#

Get external system ID for a NexusLIMS user.

Parameters:

nexuslims_username – Username from session_log.user
external_system – Target external system

Returns:

External ID if found, None otherwise

Return type:

str | None

Examples:

>>> from nexusLIMS.db.models import get_external_id
>>> from nexusLIMS.db.enums import ExternalSystem
>>> uid = get_external_id('jsmith', ExternalSystem.LABARCHIVES_ELN)
>>> print(uid)
'285489257Ho...'

nexusLIMS.db.models.get_nexuslims_username(external_id: str, external_system: ExternalSystem) → str | None[source]#

Reverse lookup: find NexusLIMS username from external ID.

Useful for harvesters that receive external IDs (e.g., NEMO user IDs) and need to map them to NexusLIMS usernames for session_log entries.

Parameters:

external_id – ID in external system
external_system – Source external system

Returns:

NexusLIMS username if found, None otherwise

Return type:

str | None

Examples:

>>> from nexusLIMS.db.models import get_nexuslims_username
>>> from nexusLIMS.db.enums import ExternalSystem
>>> username = get_nexuslims_username('12345', ExternalSystem.NEMO)
>>> print(username)
'jsmith'

nexusLIMS.db.models.store_external_id(nexuslims_username: str, external_system: ExternalSystem, external_id: str, email: str | None = None, notes: str | None = None) → ExternalUserIdentifier[source]#

Store or update external ID mapping.

If mapping exists for this user/system combination, updates it and refreshes last_verified_at. Otherwise, creates new mapping.

Parameters:

nexuslims_username – Username from session_log.user
external_system – Target external system
external_id – ID in external system
email – Optional email for verification
notes – Optional notes about this mapping

Returns:

Created or updated ExternalUserIdentifier record

Return type:

ExternalUserIdentifier

Examples:

>>> from nexusLIMS.db.models import store_external_id
>>> from nexusLIMS.db.enums import ExternalSystem
>>> record = store_external_id(
...     nexuslims_username='jsmith',
...     external_system=ExternalSystem.LABARCHIVES_ELN,
...     external_id='285489257Ho...',
...     email='jsmith@upenn.edu',
...     notes='OAuth registration portal 2026-01-25'
... )

nexusLIMS.db.models.get_all_external_ids(nexuslims_username: str) → dict[str, str][source]#

Get all external IDs for a user.

Returns dict mapping external system name to external ID. Useful for debugging or user profile displays.

Parameters:: nexuslims_username – Username from session_log.user
Returns:: Dict mapping external system name to external ID
Return type:: dict[str, str]

Examples:

>>> from nexusLIMS.db.models import get_all_external_ids
>>> ids = get_all_external_ids('jsmith')
>>> print(ids)
{
    'nemo': '12345',
    'labarchives_eln': '285489257Ho...',
    'cdcs': 'jsmith@upenn.edu'
}

nexusLIMS.db.models#

Module Contents#

Classes#

Functions#

API#

This Page

`nexusLIMS.db.models`#