agent0.chainsync.db.base

Generic database utilities

Submodules

Classes

TableWithBlockNumber

An abstract table that has block_number

DBAddrToUsername

Maps an address to a username.

DBBase

Base class to subclass from to define the schema

Functions

add_addr_to_username(→ None)

Add username mapping to postgres during agent initialization.

close_session(→ None)

Close the session.

drop_table(→ None)

Drop a table from the database.

get_addr_to_username(→ pandas.DataFrame)

Get all usermapping and returns as a pandas dataframe.

get_latest_block_number_from_table(→ int)

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

initialize_engine(→ sqlalchemy.Engine)

Initialize the postgres engine from config.

initialize_session(→ sqlalchemy.orm.Session)

Initialize the postgres session.

query_tables(→ list[str])

Return a list of tables in the database.

Package Contents

class agent0.chainsync.db.base.TableWithBlockNumber

Bases: agent0.chainsync.db.base.schema.DBBase

An abstract table that has block_number

block_number() → sqlalchemy.Column

Stubbed block_number column.

Returns:

The sqlalchemy Column object for the block number

Return type:

Column

agent0.chainsync.db.base.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.

Parameters:

  • username (str) – The logical username to attach to the wallet address.

  • addresses (list[str] | str) – A single or list of wallet addresses to map to the username.

  • session (Session) – The initialized session object.

  • user_suffix (str) – An optional suffix to add to the username mapping.

  • force_update (bool) – If true and an existing mapping is found, will overwrite.

agent0.chainsync.db.base.close_session(session: sqlalchemy.orm.Session) → None

Close the session.

Parameters:

session (Session) – The initialized session object

agent0.chainsync.db.base.drop_table(session: sqlalchemy.orm.Session, table_name: str) → None

Drop a table from the database.

Parameters:

  • session (Session) – The initialized session object

  • table_name (str) – The name of the table to be dropped

agent0.chainsync.db.base.get_addr_to_username(session: sqlalchemy.orm.Session, address: str | None = None) → pandas.DataFrame

Get all usermapping and returns as a pandas dataframe.

Parameters:

  • session (Session) – The initialized session object

  • address (str | None, optional) – The wallet address to filter the results on. Return all if None

Returns:

A DataFrame that consists of the queried pool config data

Return type:

DataFrame

agent0.chainsync.db.base.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.

Parameters:

  • table_obj (Type[Base]) – The sqlalchemy class that contains the block_number column

  • session (Session) – The initialized session object

Returns:

The latest block number from the specified table

Return type:

int

agent0.chainsync.db.base.initialize_engine(postgres_config: agent0.chainsync.PostgresConfig | None = None, ensure_database_created: bool = False) → sqlalchemy.Engine

Initialize the postgres engine from config.

Parameters:

  • postgres_config (PostgresConfig | None, optional) – The postgres config. If none, will set from .env file or set to defaults.

  • ensure_database_created (bool, optional) – If true, will create the database within postgres if it doesn’t exist. Defaults to false.

Returns:

The initialized engine object connected to postgres

Return type:

Engine

agent0.chainsync.db.base.initialize_session(postgres_config: agent0.chainsync.PostgresConfig | None = None, drop: bool = False, ensure_database_created: bool = False) → sqlalchemy.orm.Session

Initialize the postgres session.

Parameters:

  • postgres_config (PostgresConfig | None, optional) – The postgres config. If none, will set from .env file or set to defaults.

  • drop (bool, optional) – If true, will drop all tables in the database before doing anything for debugging. Defaults to false.

  • ensure_database_created (bool, optional) – If true, will create the database within postgres if it doesn’t exist. Defaults to false.

Returns:

The initialized session object

Return type:

Session

agent0.chainsync.db.base.query_tables(session: sqlalchemy.orm.Session) → list[str]

Return a list of tables in the database.

Parameters:

session (Session) – The initialized session object

Returns:

A list of table names in the database

Return type:

list[str]

class agent0.chainsync.db.base.DBAddrToUsername

Bases: DBBase

Maps an address to a username.

address: sqlalchemy.orm.Mapped[str]

The wallet address

username: sqlalchemy.orm.Mapped[str]

The logical username

class agent0.chainsync.db.base.DBBase

Bases: sqlalchemy.orm.MappedAsDataclass, sqlalchemy.orm.DeclarativeBase

Base class to subclass from to define the schema