A class for managing PostgreSQL databases across different environments.
This class provides methods for transferring data between cloud and Docker databases.
The constructor for the PostgresManager class.
Parameters:
Source code in dev_tool/db/postgres/manager.py
| def __init__(
self,
docker_database: DockerPostgresDatabase,
cloud_database: CloudPostgresDatabase,
) -> None:
"""
The constructor for the PostgresManager class.
:param docker_database: The Docker PostgreSQL database instance.
:param cloud_database: The cloud PostgreSQL database instance.
"""
self.docker_database = docker_database
self.cloud_database = cloud_database
|
docker_database = docker_database
instance-attribute
cloud_database = cloud_database
instance-attribute
cleanup
A method that cleans up temporary files after a database operation.
Source code in dev_tool/db/postgres/manager.py
| def cleanup(self) -> None:
"""A method that cleans up temporary files after a database operation."""
mount = BASE / 'mounted'
sql_input_file = mount / (self.cloud_database.config.name + '.sql')
sql_output_file = mount / (self.cloud_database.config.name + '-clean.sql')
input_destination = BASE / sql_input_file.name
if sql_input_file.exists():
if input_destination.exists():
input_destination.unlink()
sql_input_file.rename(input_destination)
output_destination = BASE / sql_output_file.name
if sql_output_file.exists():
if output_destination.exists():
output_destination.unlink()
sql_output_file.rename(output_destination)
if mount.exists():
shutil.rmtree(mount)
|
convert_directory_dump_to_sql
A method that converts a directory dump to an SQL file.
Parameters:
-
mount
(Path)
–
The path to the mounted directory.
-
sql
(Path)
–
The path to the output SQL file.
Source code in dev_tool/db/postgres/manager.py
| def convert_directory_dump_to_sql(self, mount: Path, sql: Path) -> None:
"""
A method that converts a directory dump to an SQL file.
:param mount: The path to the mounted directory.
:param sql: The path to the output SQL file.
"""
version = CONTEXT.configuration.get_docker_config().get('postgres-version')
command = [
'docker', 'run',
'--rm',
'-v', f'{mount.parent}:/tmp',
f'postgres:{version}',
'pg_restore',
'-F', 'd',
'-f', f'/tmp/mounted/{sql.name}', # noqa: S108
'--jobs=8',
'--clean',
'--if-exists',
'--verbose',
'/tmp/mounted/dump' # noqa: S108
]
subprocess.run(command, check=True)
|
from_cloud_database_to_docker_database
A method that dumps a cloud database and restores it to a Docker database.
Source code in dev_tool/db/postgres/manager.py
| def from_cloud_database_to_docker_database(self) -> None:
"""A method that dumps a cloud database and restores it to a Docker database."""
mount = BASE / 'mounted'
mount.mkdir(exist_ok=True, parents=True)
sql_input_file = BASE / 'mounted' / (self.cloud_database.config.name + '.sql')
sql_output_file = BASE / 'mounted' / (self.cloud_database.config.name + '-clean.sql')
if not mount.exists() or not sql_input_file.exists():
self.cloud_database.dump_database(mount)
if not mount.exists():
message = f'The dump directory {mount} was not saved successfully.'
raise FileNotFoundError(message)
self.convert_directory_dump_to_sql(mount, sql_input_file)
self.update_sql_file(sql_input_file, sql_output_file)
self.docker_database.disconnect_all()
if not self.docker_database.user_exists():
self.docker_database.create_user()
self.docker_database.grant_all_privileges()
self.docker_database.restore_database_from_file(sql_output_file)
self.cleanup()
|
from_decrypted_content_to_docker_database
A method that restores a Docker database from decrypted content.
Source code in dev_tool/db/postgres/manager.py
| def from_decrypted_content_to_docker_database(self, content: str) -> None:
"""A method that restores a Docker database from decrypted content."""
dump, pgcron = self._split_content_by_marker(content)
self.docker_database.disconnect_all()
if not self.docker_database.user_exists():
self.docker_database.create_user()
self.docker_database.grant_all_privileges()
self.docker_database.restore_database_from_content(dump)
if pgcron:
self.docker_database.execute_pgcron_setup(pgcron)
|
from_sql_file_to_docker_database
A method that restores a Docker database from a .sql file.
Source code in dev_tool/db/postgres/manager.py
| def from_sql_file_to_docker_database(self) -> None:
"""A method that restores a Docker database from a .sql file."""
sql_input_file = BASE / (self.cloud_database.config.name + '.sql')
sql_output_file = BASE / (self.cloud_database.config.name + '-clean.sql')
if not sql_input_file.exists():
message = f'The SQL file {sql_input_file} does not exist. Skipping...'
CONTEXT.notification.warning_text(message)
log.debug(message)
return
self.update_sql_file(sql_input_file, sql_output_file)
self.docker_database.disconnect_all()
if not self.docker_database.user_exists():
self.docker_database.create_user()
self.docker_database.grant_all_privileges()
self.docker_database.restore_database_from_file(sql_output_file)
self.cleanup()
|
update_sql_file
A method that updates references in an SQL file.
Parameters:
-
input_file
(Path)
–
The path to the input SQL file.
-
output_file
(Path)
–
The path to the output SQL file.
Source code in dev_tool/db/postgres/manager.py
| def update_sql_file(self, input_file: Path, output_file: Path) -> None:
"""
A method that updates references in an SQL file.
:param input_file: The path to the input SQL file.
:param output_file: The path to the output SQL file.
"""
with open(input_file, 'r', encoding='utf-8') as handle:
content = handle.read()
processed = self._process_content(content)
with open(output_file, 'w', encoding='utf-8') as handle:
handle.write(processed)
|