DB

class MelodieInfra.db.db.DBConn(db_name: str, db_type: str = 'sqlite', conn_params: Dict[str, str] | None = None, conn_string='')

Bases: object

DBConn provides API to write to/read from the database.

Parameters:

  • db_name – Name of database file.

  • db_type – Type of database, currently only support “sqlite”.

  • conn_params – A dict for connection parameters.

table_dtypes: Dict[str, Dict[str, str | Type[str] | Type[float] | Type[int] | Type[complex] | Type[bool] | Type[object]]] = {}
existing_connections: Dict[str, DBConn] = {}
SCENARIO_TABLE = 'simulator_scenarios'
ENVIRONMENT_RESULT_TABLE = 'environment_result'
static from_connection_string(conn_string: str) → DBConn

Create from connection string.

Parameters:

conn_string –

Returns:

get_engine()

Get the connection

Returns:

Database engine

create_connection(database_name) → Engine

Create a connection to the sqlite database.

Parameters:

database_name – Name of sqlite database file.

Returns:

classmethod create_from_existing_db(type: str, meta: Dict)

Create from existing database

Parameters:

  • type – Only support ‘sqlite’ now.

  • meta – A dict like {'path': 'path-to-sqlite'}

Returns:

Database engine.

classmethod register_dtypes(table_name: str, dtypes: Dict[str, str | Type[str] | Type[float] | Type[int] | Type[complex] | Type[bool] | Type[object]])

Register data types of a table for sqlalchemy.

classmethod get_table_dtypes(table_name: str) → Dict[str, str | Type[str] | Type[float] | Type[int] | Type[complex] | Type[bool] | Type[object]]

Get the data type of a table. If table data type is not specified, return an empty dict.

Parameters:

table_name – Name of table in database.

Returns:

close()

Close DB connection.

clear_database()

Clear the database, deleting all tables.

write_dataframe(table_name: str, data_frame: pd.DataFrame | TableBase, data_types: Dict[str, str | Type[str] | Type[float] | Type[int] | Type[complex] | Type[bool] | Type[object]] | None = None, if_exists='append')

Write a dataframe to database.

Parameters:

  • table_name – Name of table.

  • data_frame – The dataframe to be written into the database.

  • data_types – The data type for columns.

  • if_exists – A string in {‘replace’, ‘fail’, ‘append’}.

Returns:

read_dataframe(table_name: str, id_scenario: int | None = None, id_run: int | None = None, conditions: List[Tuple[str, str]] = None, df_type: str = 'pandas') → pd.DataFrame

Read a table and return all content as a dataframe.

For example:

1df = create_db_conn(config).read_dataframe('agent_params', id_scenario=0,
2                                           conditions=[('id', "<=100"), ("health_state", '=1')])
3print(df)
Parameters:

  • table_name – Name of table inside database.

  • id_scenario – Filter of scenario

  • id_run – Filter of run_id

  • conditions – Custom conditions

  • df_type – Type of dataframe, choose between “pandas” and “melodie-table”

drop_table(table_name: str)

Drop table if it exists.

Parameters:

table_name – The name of table to drop.

Returns:

query(sql) → pd.DataFrame

Execute sql command and return the result by pd.DataFrame.

Parameters:

sql – SQL phrase to execute.

Returns:

MelodieInfra.db.db.create_db_conn(config: Config) → DBConn

create a Database by current config

Returns:

DBConn object.

MelodieInfra.db.db.get_sqlite_filename(config: Config) → str

Get SQLite database filename from Melodie project config.