openpectus.aggregator.data.repository

Attributes

logger

Classes

RepositoryBase

PlotLogRepository

RecentRunRepository

RecentEngineRepository

WebPushRepository

Module Contents

openpectus.aggregator.data.repository.logger
class openpectus.aggregator.data.repository.RepositoryBase(db_session)
Parameters:

db_session (sqlalchemy.orm.Session)

db_session
class openpectus.aggregator.data.repository.PlotLogRepository(db_session)

Bases: RepositoryBase

Parameters:

db_session (sqlalchemy.orm.Session)

create_plot_log(engine_data, run_id)
Parameters:
store_new_tag_info(engine_id, run_id, tag)
Parameters:

  • engine_id (str)

  • run_id (str)

  • tag (openpectus.aggregator.models.TagValue)

get_plot_log(run_id)
Parameters:

run_id (str)

Return type:

openpectus.aggregator.data.models.PlotLog | None

get_plot_log_entry(engine_id, run_id, tag)
Parameters:

  • engine_id (str)

  • run_id (str)

  • tag (openpectus.aggregator.models.TagValue)

Return type:

openpectus.aggregator.data.models.PlotLogEntry | None

get_plot_log_entries(engine_id, run_id)
Parameters:

  • engine_id (str)

  • run_id (str)

Return type:

Iterable[openpectus.aggregator.data.models.PlotLogEntry]

store_tag_values(engine_id, run_id, tags)
Parameters:

  • engine_id (str)

  • run_id (str)

  • tags (list[openpectus.aggregator.models.TagValue])

class openpectus.aggregator.data.repository.RecentRunRepository(db_session)

Bases: RepositoryBase

Parameters:

db_session (sqlalchemy.orm.Session)

store_recent_run(engine_data, archive=None, archive_filename=None)

Store a recent run. Requires that engine_data contain run_data.

Parameters:
get_by_run_id(run_id)
Parameters:

run_id (str)

Return type:

openpectus.aggregator.data.models.RecentRun | None

get_by_engine_id(engine_id)
Parameters:

engine_id (str)

Return type:

Iterable[openpectus.aggregator.data.models.RecentRun]

get_all()
Return type:

Iterable[openpectus.aggregator.data.models.RecentRun]

get_method_and_state_by_run_id(run_id)
Parameters:

run_id (str)

get_plot_configuration_by_run_id(run_id)
Parameters:

run_id (str)

get_run_log_by_run_id(run_id)
Parameters:

run_id (str)

get_error_log_by_run_id(run_id)
Parameters:

run_id (str)

class openpectus.aggregator.data.repository.RecentEngineRepository(db_session)

Bases: RepositoryBase

Parameters:

db_session (sqlalchemy.orm.Session)

get_recent_engines()
Return type:

Sequence[openpectus.aggregator.data.models.RecentEngine]

get_recent_engine_by_engine_id(engine_id)
Parameters:

engine_id (str)

Return type:

openpectus.aggregator.data.models.RecentEngine | None

store_recent_engine(engine_data)
Parameters:

engine_data (openpectus.aggregator.models.EngineData)

class openpectus.aggregator.data.repository.WebPushRepository(db_session)

Bases: RepositoryBase

Parameters:

db_session (sqlalchemy.orm.Session)

get_notification_preferences_for_user(user_id)
Parameters:

user_id (str)

Return type:

openpectus.aggregator.data.models.WebPushNotificationPreferences | None

get_subscriptions(user_ids)
Parameters:

user_ids (list[str])

get_subscriptions_for_user(user_id)
Parameters:

user_id (str)

get_notification_preferences_for_topic(topic)
Parameters:

topic (openpectus.aggregator.models.NotificationTopic)

store_notifications_preferences(agg_notification_preferences)
Parameters:

agg_notification_preferences (openpectus.aggregator.models.WebPushNotificationPreferences)

store_subscription(agg_subscription, user_id)
Parameters:

  • agg_subscription (webpush.WebPushSubscription)

  • user_id (str)

delete_subscription(subscription)
Parameters:

subscription (openpectus.aggregator.data.models.WebPushSubscription)