docker_db package

Submodules

docker_db.dbs.cassandra_db module

docker_db.dbs.mongo_db module

MongoDB container management module.

This module provides classes to manage MongoDB database containers using Docker. It enables developers to easily create, configure, start, stop, and delete MongoDB containers for development and testing purposes.

The module defines two main classes: - MongoDBConfig: Configuration settings for MongoDB containers - MongoDB: Manager for MongoDB container lifecycle

Examples

>>> from docker_db.mongodb import MongoDBConfig, MongoDB
>>> config = MongoDBConfig(
...     user="testuser",
...     password="testpass",
...     database="testdb",
...     root_username="admin",
...     root_password="adminpass",
...     container_name="test-mongodb"
... )
>>> db = MongoDB(config)
>>> db.create_db()
>>> # Use the database...
>>> db.stop_db()

class docker_db.dbs.mongo_db.MongoDB(config)

Bases: ContainerManager

Manages lifecycle of a MongoDB container via Docker SDK.

This class provides functionality to create, start, stop, and delete MongoDB containers using the Docker SDK. It also handles database creation, user management, and connection establishment.

Parameters:: config (MongoDBConfig) – Configuration object containing MongoDB and container settings.

config

The configuration object for this MongoDB instance.

Type:: MongoDBConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

database_created

Flag indicating whether the database has been created successfully.

Type:: bool

Raises:: AssertionError – If Docker is not running when initializing.

property connection

Establish a new MongoDB connection.

Returns:: connection – A new connection to the MongoDB database.
Return type:: pymongo.MongoClient

Notes

This creates a new connection each time it’s called. The connection uses the user credentials specified in the configuration.

connection_string(db_name=None)

Get MongoDB connection string with user credentials.

Parameters:: db_name (str) – Name of the database to connect to. If None, connects to the default database.
Returns:: A connection string for MongoDB using user credentials.
Return type:: str

class docker_db.dbs.mongo_db.MongoDBConfig(**data)

Bases: ContainerConfig

Configuration for MongoDB container.

This class extends ContainerConfig with MongoDB-specific configuration options. It provides the necessary settings to create and connect to a MongoDB database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

root_password: str

root_username: str

user: str

docker_db.dbs.mssql_db module

Microsoft SQL Server container management module.

This module provides classes to manage Microsoft SQL Server database containers using Docker. It enables developers to easily create, configure, start, stop, and delete MSSQL containers for development and testing purposes.

The module defines two main classes: - MSSQLConfig: Configuration settings for Microsoft SQL Server containers - MSSQLDB: Manager for MSSQL container lifecycle

Examples

>>> from docker_db.mssql import MSSQLConfig, MSSQLDB
>>> config = MSSQLConfig(
...     user="testuser",
...     password="testpass",
...     database="testdb",
...     sa_password="StrongPassword123!",
...     container_name="test-mssql"
... )
>>> db = MSSQLDB(config)
>>> db.create_db()
>>> # Use the database...
>>> db.stop_db()

class docker_db.dbs.mssql_db.MSSQLConfig(**data)

Bases: ContainerConfig

Configuration for Microsoft SQL Server container.

This class extends ContainerConfig with MSSQL-specific configuration options. It provides the necessary settings to create and connect to a Microsoft SQL Server database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

sa_password: str

user: str

class docker_db.dbs.mssql_db.MSSQLDB(config)

Bases: ContainerManager

Manages lifecycle of a Microsoft SQL Server container via Docker SDK.

This class provides functionality to create, start, stop, and delete Microsoft SQL Server containers using the Docker SDK. It also handles database creation, user management, and connection establishment.

Parameters:: config (MSSQLConfig) – Configuration object containing SQL Server and container settings.

config

The configuration object for this SQL Server instance.

Type:: MSSQLConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

database_created

Flag indicating whether the database has been created successfully.

Type:: bool

Raises:: AssertionError – If Docker is not running when initializing.

config: MSSQLConfig

property connection

Establish a new pyodbc connection to the SQL Server database.

Returns:: connection – A new connection to the SQL Server database.
Return type:: pyodbc.Connection

Notes

This creates a new connection each time it’s called. If the database has been created (indicated by the database_created attribute), the connection will include the database name in the connection string.

connection_string(db_name=None, sql_magic=False)

Get SQL Server connection string.

Parameters:

db_name (str) – Name of the database to connect to. If None, uses the default database from config or connects without specifying a database.
sql_magic (bool) – If True, formats the connection string for use with SQL magic commands (e.g., jupyter notebooks with %sql magic). Default is False.

Returns:

A connection string for SQL Server. Format depends on sql_magic parameter.

Return type:

str

docker_db.dbs.mysql_db module

MySQL container management module.

This module provides classes to manage MySQL database containers using Docker. It enables developers to easily create, configure, start, stop, and delete MySQL containers for development and testing purposes.

The module defines two main classes: - MySQLConfig: Configuration settings for MySQL containers - MySQLDB: Manager for MySQL container lifecycle

Examples

>>> from docker_db.mysql import MySQLConfig, MySQLDB
>>> config = MySQLConfig(
...     user="testuser",
...     password="testpass",
...     database="testdb",
...     root_password="rootpass",
...     container_name="test-mysql"
... )
>>> db = MySQLDB(config)
>>> db.create_db()
>>> # Use the database...
>>> db.stop_db()

class docker_db.dbs.mysql_db.MySQLConfig(**data)

Bases: ContainerConfig

Configuration for MySQL container.

This class extends ContainerConfig with MySQL-specific configuration options. It provides the necessary settings to create and connect to a MySQL database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

root_password: str

user: str

class docker_db.dbs.mysql_db.MySQLDB(config)

Bases: ContainerManager

Manages lifecycle of a MySQL container via Docker SDK.

This class provides functionality to create, start, stop, and delete MySQL containers using the Docker SDK. It also handles database creation, user management, and connection establishment.

Parameters:: config (MySQLConfig) – Configuration object containing MySQL and container settings.

config

The configuration object for this MySQL instance.

Type:: MySQLConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

database_created

Flag indicating whether the database has been created successfully.

Type:: bool

Raises:: AssertionError – If Docker is not running when initializing.

config: MySQLConfig

property connection

Establish a new mysql.connector connection.

Returns:: connection – A new connection to the MySQL database.
Return type:: mysql.connector.connection.MySQLConnection

Notes

This creates a new connection each time it’s called. If the database has been created (indicated by the database_created attribute), the connection will include the database name in the connection string.

connection_string(db_name=None, sql_magic=False)

Get MySQL connection string.

Parameters:

db_name (str) – Name of the database to connect to. If None, uses the default database from config or connects without specifying a database.
sql_magic (bool) – If True, formats the connection string for use with SQL magic commands (e.g., jupyter notebooks with %sql magic). Default is False.

Returns:

A connection string for MySQL. Format depends on sql_magic parameter.

Return type:

str

docker_db.dbs.postgres_db module

PostgreSQL container management module.

This module provides classes to manage PostgreSQL database containers using Docker. It enables developers to easily create, configure, start, stop, and delete PostgreSQL containers for development and testing purposes.

The module defines two main classes: - PostgresConfig: Configuration settings for PostgreSQL containers - PostgresDB: Manager for PostgreSQL container lifecycle

Examples

>>> from docker_db.postgres import PostgresConfig, PostgresDB
>>> config = PostgresConfig(
...     user="testuser",
...     password="testpass",
...     database="testdb",
...     container_name="test-postgres"
... )
>>> db = PostgresDB(config)
>>> db.create_db()
>>> # Connect to the database
>>> conn = db.connection
>>> # Create a cursor and execute a query
>>> cursor = conn.cursor()
>>> cursor.execute("SELECT version();")
>>> version = cursor.fetchone()
>>> print(f"PostgreSQL version: {version[0]}")
>>> # Clean up
>>> cursor.close()
>>> db.stop_db()

class docker_db.dbs.postgres_db.PostgresConfig(**data)

Bases: ContainerConfig

Configuration for PostgreSQL container.

This class extends ContainerConfig with PostgreSQL-specific configuration options. It provides the necessary settings to create and connect to a PostgreSQL database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

user: str

class docker_db.dbs.postgres_db.PostgresDB(config)

Bases: ContainerManager

Manages lifecycle of a Postgres container via Docker SDK.

This class provides functionality to create, start, stop, and delete PostgreSQL containers using the Docker SDK. It also handles database creation and connection management.

Parameters:: config (PostgresConfig) – Configuration object containing PostgreSQL and container settings.

config

The configuration object for this PostgreSQL instance.

Type:: PostgresConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

config: PostgresConfig

property connection

Establish a new psycopg2 connection to the PostgreSQL database.

Returns:: connection – A new connection to the PostgreSQL database with RealDictCursor factory.
Return type:: psycopg2.extensions.connection

Notes

This creates a new connection each time it’s called.

connection_string(db_name=None, sql_magic=False)

Get PostgreSQL connection string.

Parameters:

db_name (str) – Name of the database to connect to. If None, uses the default database from config or connects without specifying a database.
sql_magic (bool) – If True, formats the connection string for use with SQL magic commands (e.g., jupyter notebooks with %sql magic). Default is False.

Returns:

A connection string for PostgreSQL. Format depends on sql_magic parameter.

Return type:

str

docker_db.docker.manager module

Container manager for Docker-based services.

class docker_db.docker.manager.ContainerManager(config)

Bases: object

Manages lifecycle of a PostgreSQL container via Docker SDK.

This class handles creating, starting, stopping, and monitoring a PostgreSQL container using the Docker SDK. It is designed to be subclassed with implementations for specific database connection methods.

Parameters:: config (ContainerConfig) – Configuration object containing settings for the container
Raises:: ConnectionError – If Docker daemon is not accessible

config: ContainerConfig

property connection

Establish a new psycopg2 connection to the database.

Returns:: connection – A connection to the PostgreSQL database
Return type:: psycopg2.connection
Raises:: NotImplementedError – This method must be implemented by subclasses

create_db(db_name=None, container=None, exists_ok=True, running_ok=True, force=False, retry=True)

Create the container, the database and have it running.

Parameters:

db_name (str) – Name of the database to create
container (Container) – Container reference, fetches by name if not provided
exists_ok (bool) – If True, continue if the database already exists
running_ok (bool) – If True, continue if the container is already running
force (bool) – If True, remove existing container and create a new one

Returns:

container – The container running the database

Return type:

docker.models.containers.Container

Raises:

NotImplementedError – This method must be implemented by subclasses
RuntimeError – If container creation or starting fails
ConnectionError – If database does not become ready within the configured timeout

delete_db(container=None, running_ok=False)

Delete the PostgreSQL container.

This method removes the container completely.

Parameters:

container (Container | None) – Container to delete. If omitted, resolves by configured container name.
running_ok (bool) – If True, allows deletion of a running container by stopping it first.

Returns:

Removed container object, or None when container did not exist.

Return type:

docker.models.containers.Container or None

restart_db(container=None, wait_timeout=30)

Restart the database container, ensuring it’s fully stopped before restarting.

Parameters:

wait_timeout (int) – Timeout in seconds to wait for the container to stop, by default 30

Returns:

container – The restarted container object

Return type:

docker.models.containers.Container

Raises:

RuntimeError – If container not found, fails to stop, or fails to restart
ConnectionError – If database does not become ready within the configured timeout

start_db(container=None, running_ok=True, retry=True, force=False)

Start the database container and wait until it’s healthy.

Parameters:

container (Container) – Container to start, fetches by name if not provided
running_ok (bool) – If True, continue if the container is already running and just wait for the database to be ready
force (bool) – If True, recreate the container even if it already exists

Returns:

container – The started container

Return type:

docker.models.containers.Container

Raises:

RuntimeError – If container not found or fails to start
ConnectionError – If database does not become ready within the configured timeout

state(container=None)

Get the current state of the container.

Parameters:: container (Container) – Container to check, fetches by name if not provided
Returns:: Current state of the container (“running”, “exited”, etc.)
Return type:: str

stop_db(container=None, force=False)

Stop the PostgreSQL container.

This method stops the container and prints its state.

Parameters:

container (Container | None) – Container to stop. If omitted, resolves by configured container name.
force (bool) – If True, force-stop if graceful stop does not complete.

Returns:

Stopped container object, or None if no container existed.

Return type:

docker.models.containers.Container or None

test_connection()

Ensure DB is reachable, otherwise build & start.

This method attempts to connect to the database and if unsuccessful, builds and starts a new container.

This is the main entry point for typical usage, as it handles checking for an existing database and setting up a new one if needed.

Raises:: ConnectionError – If the database is unreachable and Docker container needs to be started

docker_db.docker.model module

Container data models and defaults.

class docker_db.docker.model.ContainerConfig(**data)

Bases: BaseModel

Configuration for a Docker container running a database.

container_name: str | None

delay: int

dockerfile_path: Path | None

extra_hosts: dict[str, str] | None

host: str

image_name: str | None

init_script: Path | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

network_mode: str | None

port: int | None

project_name: str

retries: int

volume_path: Path | None

workdir: Path | None

docker_db.docker.utils module

Docker port management utilities.

This module provides utilities for managing Docker containers and network ports, including checking Docker daemon availability, port availability, and clearing ports by stopping containers.

Functions

is_docker_running : Check if Docker daemon is accessible is_port_free : Check if a port is available for binding clear_port : Stop containers using a specific port to free it up

Dependencies

docker: Docker SDK for Python
requests: HTTP library (used by docker SDK)
socket: Built-in networking interface
os: Operating system interface
time: Time-related functions

Example

>>> # Check if Docker is running
>>> if is_docker_running():
...     print("Docker is available")
...
>>> # Check if port is free
>>> if is_port_free('localhost', 8080):
...     print("Port 8080 is available")
...
>>> # Clear port by stopping containers
>>> clear_port(8080, 'my-app-')

docker_db.docker.utils.clear_port(port, container_prefix)

Clear a port by stopping containers that are using it.

Continuously monitors the specified port and stops any containers whose names start with the given prefix if they are using the port. Waits until the port is completely free before returning.

Parameters:

port (int) – Port number to clear (1-65535)
container_prefix (str) – Prefix of container names to stop. Only containers whose names start with this prefix will be stopped

Return type:

None

Notes

Checks both running and stopped containers for port usage
Only stops containers matching the name prefix
Polls every 0.5 seconds until port is free
Adds 2-second delay after clearing for system cleanup
Uses TCP protocol for port checking

Raises:: docker.errors.DockerException – If Docker daemon is not accessible

Examples

>>> # Stop all containers starting with 'webapp-' that use port 8080
>>> clear_port(8080, 'webapp-')
>>>
>>> # Stop containers starting with 'db-' that use port 5432
>>> clear_port(5432, 'db-')

docker_db.docker.utils.is_docker_running(docker_base_url=None, timeout=10)

Check if Docker engine is running and accessible.

Automatically detects the appropriate Docker socket URL based on the operating system if not provided. Tests connectivity by pinging the Docker daemon.

Parameters:

docker_base_url (str) – URL to Docker socket. If None, auto-detects based on OS: - Windows: ‘npipe:////./pipe/docker_engine’ - Unix-like: ‘unix://var/run/docker.sock’
timeout (int) – Timeout in seconds for Docker connection attempts

Returns:

True if Docker daemon is running and accessible

Return type:

bool

Raises:

ConnectionError – If Docker daemon is not accessible, with details about the connection failure

Examples

>>> is_docker_running()
True
>>> is_docker_running(timeout=5)
True
>>> is_docker_running('unix:///custom/docker.sock')
True

docker_db.docker.utils.is_port_free(host, port)

Check if a port is free for binding.

Performs two checks: 1. Tests if a socket can bind to the port 2. Verifies no running Docker containers are using the port

Parameters:

host (str) – Host address to check (e.g., ‘localhost’, ‘0.0.0.0’, ‘127.0.0.1’)
port (int) – Port number to check (1-65535)

Returns:

True if port is free (socket can bind AND no running containers use it), False otherwise

Return type:

bool

Notes

If Docker is not available, only performs socket binding test
Only checks running containers, not stopped ones
Uses TCP protocol for both socket and container checks

Examples

>>> is_port_free('localhost', 8080)
True
>>> is_port_free('0.0.0.0', 80)
False

Module contents

Top-level package exports.

class docker_db.ContainerConfig(**data)

Bases: BaseModel

Configuration for a Docker container running a database.

container_name: str | None

delay: int

dockerfile_path: Path | None

extra_hosts: dict[str, str] | None

host: str

image_name: str | None

init_script: Path | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

network_mode: str | None

port: int | None

project_name: str

retries: int

volume_path: Path | None

workdir: Path | None

class docker_db.ContainerManager(config)

Bases: object

Manages lifecycle of a PostgreSQL container via Docker SDK.

This class handles creating, starting, stopping, and monitoring a PostgreSQL container using the Docker SDK. It is designed to be subclassed with implementations for specific database connection methods.

Parameters:: config (ContainerConfig) – Configuration object containing settings for the container
Raises:: ConnectionError – If Docker daemon is not accessible

config: ContainerConfig

property connection

Establish a new psycopg2 connection to the database.

Returns:: connection – A connection to the PostgreSQL database
Return type:: psycopg2.connection
Raises:: NotImplementedError – This method must be implemented by subclasses

create_db(db_name=None, container=None, exists_ok=True, running_ok=True, force=False, retry=True)

Create the container, the database and have it running.

Parameters:

db_name (str) – Name of the database to create
container (Container) – Container reference, fetches by name if not provided
exists_ok (bool) – If True, continue if the database already exists
running_ok (bool) – If True, continue if the container is already running
force (bool) – If True, remove existing container and create a new one

Returns:

container – The container running the database

Return type:

docker.models.containers.Container

Raises:

NotImplementedError – This method must be implemented by subclasses
RuntimeError – If container creation or starting fails
ConnectionError – If database does not become ready within the configured timeout

delete_db(container=None, running_ok=False)

Delete the PostgreSQL container.

This method removes the container completely.

Parameters:

container (Container | None) – Container to delete. If omitted, resolves by configured container name.
running_ok (bool) – If True, allows deletion of a running container by stopping it first.

Returns:

Removed container object, or None when container did not exist.

Return type:

docker.models.containers.Container or None

restart_db(container=None, wait_timeout=30)

Restart the database container, ensuring it’s fully stopped before restarting.

Parameters:

wait_timeout (int) – Timeout in seconds to wait for the container to stop, by default 30

Returns:

container – The restarted container object

Return type:

docker.models.containers.Container

Raises:

RuntimeError – If container not found, fails to stop, or fails to restart
ConnectionError – If database does not become ready within the configured timeout

start_db(container=None, running_ok=True, retry=True, force=False)

Start the database container and wait until it’s healthy.

Parameters:

container (Container) – Container to start, fetches by name if not provided
running_ok (bool) – If True, continue if the container is already running and just wait for the database to be ready
force (bool) – If True, recreate the container even if it already exists

Returns:

container – The started container

Return type:

docker.models.containers.Container

Raises:

RuntimeError – If container not found or fails to start
ConnectionError – If database does not become ready within the configured timeout

state(container=None)

Get the current state of the container.

Parameters:: container (Container) – Container to check, fetches by name if not provided
Returns:: Current state of the container (“running”, “exited”, etc.)
Return type:: str

stop_db(container=None, force=False)

Stop the PostgreSQL container.

This method stops the container and prints its state.

Parameters:

container (Container | None) – Container to stop. If omitted, resolves by configured container name.
force (bool) – If True, force-stop if graceful stop does not complete.

Returns:

Stopped container object, or None if no container existed.

Return type:

docker.models.containers.Container or None

test_connection()

Ensure DB is reachable, otherwise build & start.

This method attempts to connect to the database and if unsuccessful, builds and starts a new container.

This is the main entry point for typical usage, as it handles checking for an existing database and setting up a new one if needed.

Raises:: ConnectionError – If the database is unreachable and Docker container needs to be started

class docker_db.MSSQLConfig(**data)

Bases: ContainerConfig

Configuration for Microsoft SQL Server container.

This class extends ContainerConfig with MSSQL-specific configuration options. It provides the necessary settings to create and connect to a Microsoft SQL Server database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

sa_password: str

user: str

class docker_db.MSSQLDB(config)

Bases: ContainerManager

Manages lifecycle of a Microsoft SQL Server container via Docker SDK.

This class provides functionality to create, start, stop, and delete Microsoft SQL Server containers using the Docker SDK. It also handles database creation, user management, and connection establishment.

Parameters:: config (MSSQLConfig) – Configuration object containing SQL Server and container settings.

config

The configuration object for this SQL Server instance.

Type:: MSSQLConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

database_created

Flag indicating whether the database has been created successfully.

Type:: bool

Raises:: AssertionError – If Docker is not running when initializing.

config: MSSQLConfig

property connection

Establish a new pyodbc connection to the SQL Server database.

Returns:: connection – A new connection to the SQL Server database.
Return type:: pyodbc.Connection

Notes

This creates a new connection each time it’s called. If the database has been created (indicated by the database_created attribute), the connection will include the database name in the connection string.

connection_string(db_name=None, sql_magic=False)

Get SQL Server connection string.

Parameters:

db_name (str) – Name of the database to connect to. If None, uses the default database from config or connects without specifying a database.
sql_magic (bool) – If True, formats the connection string for use with SQL magic commands (e.g., jupyter notebooks with %sql magic). Default is False.

Returns:

A connection string for SQL Server. Format depends on sql_magic parameter.

Return type:

str

class docker_db.MongoDB(config)

Bases: ContainerManager

Manages lifecycle of a MongoDB container via Docker SDK.

This class provides functionality to create, start, stop, and delete MongoDB containers using the Docker SDK. It also handles database creation, user management, and connection establishment.

Parameters:: config (MongoDBConfig) – Configuration object containing MongoDB and container settings.

config

The configuration object for this MongoDB instance.

Type:: MongoDBConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

database_created

Flag indicating whether the database has been created successfully.

Type:: bool

Raises:: AssertionError – If Docker is not running when initializing.

config: MongoDBConfig

property connection

Establish a new MongoDB connection.

Returns:: connection – A new connection to the MongoDB database.
Return type:: pymongo.MongoClient

Notes

This creates a new connection each time it’s called. The connection uses the user credentials specified in the configuration.

connection_string(db_name=None)

Get MongoDB connection string with user credentials.

Parameters:: db_name (str) – Name of the database to connect to. If None, connects to the default database.
Returns:: A connection string for MongoDB using user credentials.
Return type:: str

class docker_db.MongoDBConfig(**data)

Bases: ContainerConfig

Configuration for MongoDB container.

This class extends ContainerConfig with MongoDB-specific configuration options. It provides the necessary settings to create and connect to a MongoDB database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

root_password: str

root_username: str

user: str

class docker_db.MySQLConfig(**data)

Bases: ContainerConfig

Configuration for MySQL container.

This class extends ContainerConfig with MySQL-specific configuration options. It provides the necessary settings to create and connect to a MySQL database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

root_password: str

user: str

class docker_db.MySQLDB(config)

Bases: ContainerManager

Manages lifecycle of a MySQL container via Docker SDK.

This class provides functionality to create, start, stop, and delete MySQL containers using the Docker SDK. It also handles database creation, user management, and connection establishment.

Parameters:: config (MySQLConfig) – Configuration object containing MySQL and container settings.

config

The configuration object for this MySQL instance.

Type:: MySQLConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

database_created

Flag indicating whether the database has been created successfully.

Type:: bool

Raises:: AssertionError – If Docker is not running when initializing.

config: MySQLConfig

property connection

Establish a new mysql.connector connection.

Returns:: connection – A new connection to the MySQL database.
Return type:: mysql.connector.connection.MySQLConnection

Notes

This creates a new connection each time it’s called. If the database has been created (indicated by the database_created attribute), the connection will include the database name in the connection string.

connection_string(db_name=None, sql_magic=False)

Get MySQL connection string.

Parameters:

db_name (str) – Name of the database to connect to. If None, uses the default database from config or connects without specifying a database.
sql_magic (bool) – If True, formats the connection string for use with SQL magic commands (e.g., jupyter notebooks with %sql magic). Default is False.

Returns:

A connection string for MySQL. Format depends on sql_magic parameter.

Return type:

str

class docker_db.Neo4jConfig(**data)

Bases: ContainerConfig

Configuration for a Neo4j container.

database: str

env_vars: dict

http_port: int

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

user: str

class docker_db.Neo4jDB(config)

Bases: ContainerManager

Manages lifecycle of a Neo4j container via Docker SDK.

The connection property returns a neo4j.Driver using the Bolt protocol. Pass it directly to Neo4jGraphStore (LlamaIndex) or Neo4jGraph (LangChain) for GraphRAG pipelines.

Parameters:: config (Neo4jConfig) – Configuration for the Neo4j container.

config: Neo4jConfig

property connection

Return a neo4j.Driver connected via Bolt.

The neo4j package must be installed (pip install neo4j).

Returns:: An authenticated driver ready for session creation.
Return type:: neo4j.Driver

connection_string(db_name=None)

Return a Bolt URI for the configured host and port.

Return type:: str

class docker_db.OllamaConfig(**data)

Bases: ContainerConfig

Configuration for an Ollama container.

database: str

env_vars: dict

model: str | None

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

port: int

pull_model_on_create: bool

class docker_db.OllamaDB(config)

Bases: ContainerManager

Manages lifecycle of an Ollama container via Docker SDK.

property base_url: str: Return Ollama API base URL.

config: OllamaConfig

property connection

Establish a new HTTP session to the Ollama API.

Returns:: A new HTTP session object.
Return type:: requests.Session

connection_string(db_name=None, sql_magic=False)

Get Ollama base URL.

Parameters:

db_name (str) – Unused placeholder for API compatibility with other managers.
sql_magic (bool) – Unused placeholder for API compatibility with SQL-oriented managers.

Returns:

Base URL of the Ollama API endpoint.

Return type:

str

list_models()

List locally available models.

Return type:: list[dict]

pull_model(model)

Pull a model via Ollama API.

Parameters:: model (str) – Model name to download into the local Ollama store.
Returns:: JSON response returned by Ollama pull endpoint.
Return type:: dict

test_connection(): Ensure Ollama API is reachable.

class docker_db.OpenSearchConfig(**data)

Bases: ContainerConfig

Configuration for OpenSearch container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

port: int

use_bind_mount: bool

class docker_db.OpenSearchDB(config)

Bases: ContainerManager

Manages lifecycle of an OpenSearch container via Docker SDK.

config: OpenSearchConfig

property connection

Establish a new OpenSearch client connection.

Returns:: A client connected to the configured OpenSearch endpoint.
Return type:: OpenSearch

connection_string(db_name=None, sql_magic=False)

Get OpenSearch base URL.

Parameters:

db_name (str) – Unused for OpenSearch. Present for interface compatibility.
sql_magic (bool) – Unused for OpenSearch. Present for interface compatibility.

Returns:

Base HTTP URL for the OpenSearch endpoint.

Return type:

str

test_connection(): Ensure OpenSearch API is reachable.

class docker_db.PostgresConfig(**data)

Bases: ContainerConfig

Configuration for PostgreSQL container.

This class extends ContainerConfig with PostgreSQL-specific configuration options. It provides the necessary settings to create and connect to a PostgreSQL database running in a Docker container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

password: str

port: int

user: str

class docker_db.PostgresDB(config)

Bases: ContainerManager

Manages lifecycle of a Postgres container via Docker SDK.

This class provides functionality to create, start, stop, and delete PostgreSQL containers using the Docker SDK. It also handles database creation and connection management.

Parameters:: config (PostgresConfig) – Configuration object containing PostgreSQL and container settings.

config

The configuration object for this PostgreSQL instance.

Type:: PostgresConfig

client

Docker client for interacting with the Docker daemon.

Type:: docker.client.DockerClient

config: PostgresConfig

property connection

Establish a new psycopg2 connection to the PostgreSQL database.

Returns:: connection – A new connection to the PostgreSQL database with RealDictCursor factory.
Return type:: psycopg2.extensions.connection

Notes

This creates a new connection each time it’s called.

connection_string(db_name=None, sql_magic=False)

Get PostgreSQL connection string.

Parameters:

db_name (str) – Name of the database to connect to. If None, uses the default database from config or connects without specifying a database.
sql_magic (bool) – If True, formats the connection string for use with SQL magic commands (e.g., jupyter notebooks with %sql magic). Default is False.

Returns:

A connection string for PostgreSQL. Format depends on sql_magic parameter.

Return type:

str

class docker_db.QdrantConfig(**data)

Bases: ContainerConfig

Configuration for Qdrant container.

database: str

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

port: int

vector_size: int

class docker_db.QdrantDB(config)

Bases: ContainerManager

Manages lifecycle of a Qdrant container via Docker SDK.

config: QdrantConfig

property connection

Establish a new Qdrant client connection.

Returns:: A client connected to the configured Qdrant endpoint.
Return type:: QdrantClient

connection_string(db_name=None, sql_magic=False)

Get Qdrant base URL.

Parameters:

db_name (str) – Unused for Qdrant. Present for interface compatibility.
sql_magic (bool) – Unused for Qdrant. Present for interface compatibility.

Returns:

Base HTTP URL for the Qdrant endpoint.

Return type:

str

test_connection(): Ensure Qdrant API is reachable.

class docker_db.RedisConfig(**data)

Bases: ContainerConfig

Configuration for Redis container.

database: int

env_vars: dict

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(__context__)

Populate defaults derived from database type and working directory.

Parameters:: __context__ (Any) – Optional pydantic post-init context passed by the model runtime.

port: int

class docker_db.RedisDB(config)

Bases: ContainerManager

Manages lifecycle of a Redis container via Docker SDK.

config: RedisConfig

property connection

Establish a new redis-py connection.

Returns:: A Redis client connected to the configured host/port/db.
Return type:: redis.Redis

connection_string(db_name=None)

Get Redis connection string.

Parameters:: db_name (int | None) – Redis logical database index. If omitted, uses configured default.
Returns:: Redis URI for host, port, and logical database.
Return type:: str