Bases: PostgresDatabaseStrategy
A class for interacting with PostgreSQL databases in Docker containers.
This class provides methods for creating, dropping, and managing Docker PostgreSQL databases.
The constructor for the DockerPostgresDatabase class.
It initializes the database configuration from environment variables.
Source code in dev_tool/db/postgres/docker.py
| def __init__(self) -> None:
"""
The constructor for the DockerPostgresDatabase class.
It initializes the database configuration from environment variables.
"""
config = DatabaseConfig.from_env()
super().__init__(config)
|
create_database
A method that creates a database in a Docker container.
Source code in dev_tool/db/postgres/docker.py
| @override
def create_database(self) -> None:
"""A method that creates a database in a Docker container."""
connection = None
try:
connection = self._connect(autocommit=True)
with connection.cursor() as cursor:
identifier = Identifier(self.config.name)
query = SQL('CREATE DATABASE {}').format(identifier)
cursor.execute(query)
except PostgreSQLConnectionError:
raise
except Exception as exception:
message = f'Unexpected error creating database {self.config.name}: {exception}'
CONTEXT.notification.error_banner(message)
log.exception(message)
raise PostgreSQLConnectionError(message) from None
finally:
if connection:
connection.close()
|
drop_database
A method that drops a database from a Docker container.
This method first terminates all active connections to the database.
Source code in dev_tool/db/postgres/docker.py
| @override
def drop_database(self) -> None:
"""
A method that drops a database from a Docker container.
This method first terminates all active connections to the database.
"""
connection = None
try:
connection = self._connect(autocommit=True)
with connection.cursor() as cursor:
query = """
SELECT pg_terminate_backend(pg_stat_activity.pid)
FROM pg_stat_activity
WHERE pg_stat_activity.datname = %s AND pid <> pg_backend_pid()
"""
cursor.execute(
query,
(self.config.name,)
)
identifier = Identifier(self.config.name)
query = SQL('DROP DATABASE IF EXISTS {}').format(identifier)
cursor.execute(query)
except PostgreSQLConnectionError:
raise
except Exception as exception:
message = f'Unexpected error dropping database {self.config.name}: {exception}'
CONTEXT.notification.error_banner(message)
log.exception(message)
raise PostgreSQLConnectionError(message) from None
finally:
if connection:
connection.close()
|