Source code for mincepy.archive_factory
# -*- coding: utf-8 -*-
import logging
from . import historians
from . import plugins
__all__ = "create_archive", "create_historian"
logger = logging.getLogger(__name__) # pylint: disable=invalid-name
[docs]def create_archive(uri: str, connect_timeout=30000):
"""Create an archive type based on a uri string
:param uri: the specification of where to connect to
:param connect_timeout: a connection timeout (in milliseconds)
"""
from . import mongo # pylint: disable=import-outside-toplevel, cyclic-import
archive = mongo.connect(uri, timeout=connect_timeout)
logger.info("Connected to archive with uri: %s", uri)
return archive
[docs]def create_historian(
archive_uri: str, apply_plugins=True, connect_timeout=30000
) -> historians.Historian:
"""Convenience function to create a standard historian directly from an archive URI
:param archive_uri: the specification of where to connect to
:param apply_plugins: register the plugin types with the new historian
:param connect_timeout: a connection timeout (in milliseconds)
"""
historian = historians.Historian(
create_archive(archive_uri, connect_timeout=connect_timeout)
)
if apply_plugins:
historian.register_types(plugins.get_types())
return historian