swh.core.db.db_utils module

swh.core.db.db_utils.now()

swh.core.db.db_utils.stored_procedure(stored_proc)

decorator to execute remote stored procedure, specified as argument

Generally, the body of the decorated function should be empty. If it is not, the stored procedure will be executed first; the function body then.

swh.core.db.db_utils.jsonize(value)

Convert a value to a psycopg JSON object if necessary

swh.core.db.db_utils.connect_to_conninfo(db_or_conninfo: str | Connection[Any]) → Iterator[Connection[Any]]

Connect to the database passed as argument.

Parameters:

db_or_conninfo – A database connection, or a database connection info string

Returns:

a connected database handle or None if the database is not initialized

swh.core.db.db_utils.swh_db_version(db_or_conninfo: str | Connection[Any]) → int | None

Retrieve the swh version of the database.

If the database is not initialized, this logs a warning and returns None.

Parameters:

db_or_conninfo – A database connection, or a database connection info string

Returns:

Either the version of the database, or None if it couldn’t be detected

swh.core.db.db_utils.swh_db_versions(db_or_conninfo: str | Connection[Any]) → List[Tuple[int, datetime, str]] | None

Retrieve the swh version history of the database.

If the database is not initialized, this logs a warning and returns None.

Parameters:

db_or_conninfo – A database connection, or a database connection info string

Returns:

Either the version of the database, or None if it couldn’t be detected

swh.core.db.db_utils.swh_db_upgrade(conninfo: str, modname: str, to_version: int | None = None) → int

Upgrade the database at conninfo for module modname

This will run migration scripts found in the sql/upgrades subdirectory of the module modname. By default, this will upgrade to the latest declared version.

Parameters:

• conninfo – A database connection, or a database connection info string

• modname – datastore module the database stores content for

• to_version – if given, update the database to this version rather than the latest

swh.core.db.db_utils.swh_db_module(db_or_conninfo: str | Connection[Any]) → str | None

Retrieve the swh module used to create the database.

If the database is not initialized, this logs a warning and returns None.

Parameters:

db_or_conninfo – A database connection, or a database connection info string

Returns:

Either the module of the database, or None if it couldn’t be detected

swh.core.db.db_utils.swh_set_db_module(db_or_conninfo: str | Connection[Any], module: str, force=False) → None

Set the swh module used to create the database.

Fails if the dbmodule is already set or the table does not exist.

Parameters:

• db_or_conninfo – A database connection, or a database connection info string

• module – the swh module to register

swh.core.db.db_utils.swh_set_db_version(db_or_conninfo: str | Connection[Any], version: int, ts: datetime | None = None, desc: str = 'Work in progress') → None

Set the version of the database.

Fails if the dbversion table does not exists.

Parameters:

• db_or_conninfo – A database connection, or a database connection info string

• version – the version to add

swh.core.db.db_utils.swh_db_flavor(db_or_conninfo: str | Connection[Any]) → str | None

Retrieve the swh flavor of the database.

If the database is not initialized, or the database doesn’t support flavors, this returns None.

Parameters:

db_or_conninfo – A database connection, or a database connection info string

Returns:

The flavor of the database, or None if it could not be detected.

swh.core.db.db_utils.import_swhmodule(modname: str) → ModuleType | None

swh.core.db.db_utils.get_sql_for_package(modname: str, upgrade: bool = False) → List[Path]

Return the (sorted) list of sql script files for the given swh module

If upgrade is True, return the list of available migration scripts, otherwise, return the list of initialization scripts.

swh.core.db.db_utils.populate_database_for_package(modname: str, conninfo: str, flavor: str | None = None) → Tuple[bool, int | None, str | None]

Populate the database, pointed at with conninfo, using the SQL files found in the package modname. Also fill the ‘dbmodule’ table with the given modname.

Parameters:

• modname – Name of the module of which we’re loading the files

• conninfo – connection info string for the SQL database

• flavor – the module-specific flavor which we want to initialize the database under

Returns:

whether the database has been initialized; the current version of the database; if it exists, the flavor of the database.

Return type:

Tuple with three elements

swh.core.db.db_utils.initialize_database_for_module(modname: str, version: int, flavor: str | None = None, **kwargs)

Helper function to initialize and populate a database for the given module

This aims at helping the usage of pytest_postgresql for swh.core.db based datastores. Typical usage will be (here for swh.storage):

from pytest_postgresql import factories

storage_postgresql_proc = factories.postgresql_proc(
  load=[partial(initialize_database_for_module, modname="storage", version=42)]
)
storage_postgresql = factories.postgresql("storage_postgresql_proc")

swh.core.db.db_utils.get_database_info(conninfo: str | Connection[Any]) → Tuple[str | None, int | None, str | None]

Get version, flavor and module of the db

swh.core.db.db_utils.parse_dsn_or_dbname(dsn_or_dbname: str) → Dict[str, str]

Parse a psycopg dsn, falling back to supporting plain database names as well

swh.core.db.db_utils.init_admin_extensions(modname: str, conninfo: str) → None

The remaining initialization process – running -superuser- SQL files – is done using the given conninfo, thus connecting to the newly created database

swh.core.db.db_utils.create_database_for_package(modname: str, conninfo: str, template: str = 'template1')

Create the database pointed at with conninfo, and initialize it using -superuser- SQL files found in the package modname.

Parameters:

• modname – Name of the module of which we’re loading the files

• conninfo – connection info string or plain database name for the SQL database

• template – the name of the database to connect to and use as template to create the new database

swh.core.db.db_utils.dsn_with_password(con: Connection[Any])

fetch the connection info with password

swh.core.db.db_utils.execute_sqlfiles(sqlfiles: Collection[Path], db_or_conninfo: str | Connection[Any], flavor: str | None = None)

Execute a list of SQL files on the database pointed at with db_or_conninfo.

Parameters:

• sqlfiles – List of SQL files to execute

• db_or_conninfo – A database connection, or a database connection info string

• flavor – the database flavor to initialize