arrakis.utils.database ====================== .. py:module:: arrakis.utils.database .. autoapi-nested-parse:: Database utilities .. !! processed by numpydoc !! Functions --------- .. autoapisummary:: arrakis.utils.database.get_beam_inf_db arrakis.utils.database.get_db arrakis.utils.database.get_field_db arrakis.utils.database.test_db arrakis.utils.database.validate_sbid_field_pair Module Contents --------------- .. py:function:: get_beam_inf_db(host: str, epoch: int, username=None, password=None) -> pymongo.collection.Collection Get MongoDBs :Parameters: * **host** (*str*) -- Mongo host IP. * **username** (*str, optional*) -- Username. Defaults to None. * **password** (*str, optional*) -- Password. Defaults to None. :returns: beams_col, island_col, comp_col :rtype: pymongo.Collection .. !! processed by numpydoc !! .. py:function:: get_db(host: str, epoch: int, username: str | None = None, password: str | None = None) -> tuple[pymongo.collection.Collection, pymongo.collection.Collection, pymongo.collection.Collection] Get MongoDBs :Parameters: * **host** (*str*) -- Mongo host IP. * **username** (*str, optional*) -- Username. Defaults to None. * **password** (*str, optional*) -- Password. Defaults to None. :returns: beams_col, island_col, comp_col :rtype: Tuple[Collection, Collection, Collection] .. !! processed by numpydoc !! .. py:function:: get_field_db(host: str, epoch: int, username=None, password=None) -> pymongo.collection.Collection Get MongoDBs :Parameters: * **host** (*str*) -- Mongo host IP. * **username** (*str, optional*) -- Username. Defaults to None. * **password** (*str, optional*) -- Password. Defaults to None. :returns: beams_col, island_col, comp_col :rtype: pymongo.Collection .. !! processed by numpydoc !! .. py:function:: test_db(host: str, username: str | None = None, password: str | None = None) -> bool Test connection to MongoDB :Parameters: * **host** (*str*) -- Mongo host IP. * **username** (*str, optional*) -- Mongo username. Defaults to None. * **password** (*str, optional*) -- Mongo password. Defaults to None. :returns: True if connection successful :rtype: bool :raises Exception: If connection fails. .. !! processed by numpydoc !! .. py:function:: validate_sbid_field_pair(field_name: str, sbid: int, field_col: pymongo.collection.Collection) -> bool Validate field and sbid pair :Parameters: * **field_name** (*str*) -- Field name. * **sbid** (*int*) -- SBID. * **field_col** (*Collection*) -- Field collection. :raises bool: If field name and sbid pair is valid. .. !! processed by numpydoc !!