49 lines
1.5 KiB
Python
49 lines
1.5 KiB
Python
import docker
|
|
import os
|
|
|
|
|
|
class PostgresContainer():
|
|
def __init__(self):
|
|
self.postgres_user = 'locality'
|
|
self.postgres_password = 'wkyhjofg2837f'
|
|
self.postgres_db = 'locality'
|
|
|
|
def get_url(self):
|
|
return 'postgres://{}:{}@localhost/{}'.format(
|
|
self.postgres_user,
|
|
self.postgres_password,
|
|
self.postgres_db)
|
|
|
|
def __enter__(self):
|
|
postgres_env = {
|
|
'POSTGRES_USER': self.postgres_user,
|
|
'POSTGRES_PASSWORD': self.postgres_password,
|
|
'POSTGRES_DB': self.postgres_db,
|
|
}
|
|
|
|
self.docker_client = docker.from_env()
|
|
self.postgres_container = self.docker_client.containers.run(
|
|
'postgres:14.11',
|
|
environment=postgres_env,
|
|
detach=True,
|
|
ports={'5432/tcp': ('127.0.0.1', 5432)},
|
|
)
|
|
|
|
self.docker_stream = self.postgres_container.attach(stream=True)
|
|
for line_binary in self.docker_stream:
|
|
line = line_binary.decode('utf-8')
|
|
print(line)
|
|
if 'listening on IPv4 address "0.0.0.0", port 5432' in line:
|
|
break
|
|
for line_binary in self.docker_stream:
|
|
line = line_binary.decode('utf-8')
|
|
print(line)
|
|
if 'database system is ready to accept connections' in line:
|
|
break
|
|
|
|
return self
|
|
|
|
def __exit__(self, type, value, traceback):
|
|
self.postgres_container.stop()
|
|
self.postgres_container.remove()
|