agent0.chainsync.db.base.interface
==================================

.. py:module:: agent0.chainsync.db.base.interface

.. autoapi-nested-parse::

   Initialize Postgres Server.

   ..
       !! processed by numpydoc !!


Classes
-------

.. autoapisummary::

   agent0.chainsync.db.base.interface.TableWithBlockNumber


Functions
---------

.. autoapisummary::

   agent0.chainsync.db.base.interface.query_tables
   agent0.chainsync.db.base.interface.drop_table
   agent0.chainsync.db.base.interface.initialize_engine
   agent0.chainsync.db.base.interface.initialize_session
   agent0.chainsync.db.base.interface.close_session
   agent0.chainsync.db.base.interface.add_addr_to_username
   agent0.chainsync.db.base.interface.get_addr_to_username
   agent0.chainsync.db.base.interface.get_latest_block_number_from_table


Module Contents
---------------

.. py:function:: query_tables(session: sqlalchemy.orm.Session) -> list[str]

   
   Return a list of tables in the database.

   :param session: The initialized session object
   :type session: Session

   :returns: A list of table names in the database
   :rtype: list[str]















   ..
       !! processed by numpydoc !!

.. py:function:: drop_table(session: sqlalchemy.orm.Session, table_name: str) -> None

   
   Drop a table from the database.

   :param session: The initialized session object
   :type session: Session
   :param table_name: The name of the table to be dropped
   :type table_name: str















   ..
       !! processed by numpydoc !!

.. py:function:: initialize_engine(postgres_config: agent0.chainsync.PostgresConfig | None = None, ensure_database_created: bool = False) -> sqlalchemy.Engine

   
   Initialize the postgres engine from config.

   :param postgres_config: The postgres config. If none, will set from `.env` file or set to defaults.
   :type postgres_config: PostgresConfig | None, optional
   :param ensure_database_created: If true, will create the database within postgres if it doesn't exist. Defaults to false.
   :type ensure_database_created: bool, optional

   :returns: The initialized engine object connected to postgres
   :rtype: Engine















   ..
       !! processed by numpydoc !!

.. py:function:: initialize_session(postgres_config: agent0.chainsync.PostgresConfig | None = None, drop: bool = False, ensure_database_created: bool = False) -> sqlalchemy.orm.Session

   
   Initialize the postgres session.

   :param postgres_config: The postgres config. If none, will set from `.env` file or set to defaults.
   :type postgres_config: PostgresConfig | None, optional
   :param drop: If true, will drop all tables in the database before doing anything for debugging.
                Defaults to false.
   :type drop: bool, optional
   :param ensure_database_created: If true, will create the database within postgres if it doesn't exist. Defaults to false.
   :type ensure_database_created: bool, optional

   :returns: The initialized session object
   :rtype: Session















   ..
       !! processed by numpydoc !!

.. py:function:: close_session(session: sqlalchemy.orm.Session) -> None

   
   Close the session.

   :param session: The initialized session object
   :type session: Session















   ..
       !! processed by numpydoc !!

.. py:function:: add_addr_to_username(username: str, addresses: list[str] | str, session: sqlalchemy.orm.Session, user_suffix: str = '', force_update: bool = False) -> None

   
   Add username mapping to postgres during agent initialization.

   :param username: The logical username to attach to the wallet address.
   :type username: str
   :param addresses: A single or list of wallet addresses to map to the username.
   :type addresses: list[str] | str
   :param session: The initialized session object.
   :type session: Session
   :param user_suffix: An optional suffix to add to the username mapping.
   :type user_suffix: str
   :param force_update: If true and an existing mapping is found, will overwrite.
   :type force_update: bool















   ..
       !! processed by numpydoc !!

.. py:function:: get_addr_to_username(session: sqlalchemy.orm.Session, address: str | None = None) -> pandas.DataFrame

   
   Get all usermapping and returns as a pandas dataframe.

   :param session: The initialized session object
   :type session: Session
   :param address: The wallet address to filter the results on. Return all if None
   :type address: str | None, optional

   :returns: A DataFrame that consists of the queried pool config data
   :rtype: DataFrame















   ..
       !! processed by numpydoc !!

.. py:class:: TableWithBlockNumber

   Bases: :py:obj:`agent0.chainsync.db.base.schema.DBBase`


   
   An abstract table that has block_number
















   ..
       !! processed by numpydoc !!

   .. py:method:: block_number() -> sqlalchemy.Column

      
      Stubbed block_number column.

      :returns: The sqlalchemy Column object for the block number
      :rtype: Column















      ..
          !! processed by numpydoc !!


.. py:function:: get_latest_block_number_from_table(table_obj: Type[agent0.chainsync.db.base.schema.DBBase], session: sqlalchemy.orm.Session) -> int

   
   Get the latest block number based on the specified table in the db.

   :param table_obj: The sqlalchemy class that contains the block_number column
   :type table_obj: Type[Base]
   :param session: The initialized session object
   :type session: Session

   :returns: The latest block number from the specified table
   :rtype: int















   ..
       !! processed by numpydoc !!