Skip to content

docker

dev_tool.db.postgres.docker

log = logging.getLogger(__name__) module-attribute

DockerPostgresDatabase

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 
30
31
32
33
34
35
36
37
38
39
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 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@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}'
        emit_error(message)

        log.exception(message)
        raise PostgreSQLConnectionError(message) from exception
    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 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@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}'
        emit_error(message)

        log.exception(message)
        raise PostgreSQLConnectionError(message) from exception
    finally:
        if connection:
            connection.close()

dump_database

A method that dumps a Docker database to a plain SQL file.

The dump is written without ownership and privilege statements so that it can be restored into any local database.

Parameters:

  • output_file (Path) –

    The path to the output SQL file.

Source code in dev_tool/db/postgres/docker.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def dump_database(self, output_file: Path) -> None:
    """
    A method that dumps a Docker database to a plain SQL file.

    The dump is written without ownership and privilege statements so that it
    can be restored into any local database.

    :param output_file: The path to the output SQL file.
    """

    try:
        output_file.parent.mkdir(exist_ok=True, parents=True)

        version = CONTEXT.configuration.get_docker_config().get('postgres-version')

        command = [
            'docker', 'run',
            '--rm',
            '-e', f'PGPASSWORD={self.config.password}',
            f'postgres:{version}',
            'pg_dump',
            '-h', 'host.docker.internal',
            '-p', self.config.port,
            '-U', self.config.user,
            '-d', self.config.name,
            '--no-owner',
            '--no-acl'
        ]

        with output_file.open('wb') as handle:
            subprocess.run(command, check=True, stdout=handle)
    except subprocess.CalledProcessError as exception:
        output_file.unlink(missing_ok=True)

        message = f'Failed to dump database {self.config.name}: {exception}'
        emit_error(message)

        log.exception(message)
        raise PostgreSQLConnectionError(message) from exception
    except Exception as exception:
        output_file.unlink(missing_ok=True)

        message = f'Unexpected error dumping database {self.config.name}: {exception}'
        emit_error(message)

        log.exception(message)
        raise PostgreSQLConnectionError(message) from exception