Method
__init__
(
self,
port: int,
host: str,
user: str,
password: str | None,
database: str | None,
require_ssl: bool,
)
Source from the content-addressed store, hash-verified
| 76 | """An API to the database under test.""" |
| 77 | |
| 78 | def __init__( |
| 79 | self, |
| 80 | port: int, |
| 81 | host: str, |
| 82 | user: str, |
| 83 | password: str | None, |
| 84 | database: str | None, |
| 85 | require_ssl: bool, |
| 86 | ) -> None: |
| 87 | logging.debug(f"Initialize Database with host={host} port={port}, user={user}") |
| 88 | |
| 89 | self.conn = psycopg.connect( |
| 90 | host=host, |
| 91 | port=port, |
| 92 | user=user, |
| 93 | password=password, |
| 94 | dbname=database, |
| 95 | sslmode="require" if require_ssl else "disable", |
| 96 | ) |
| 97 | self.conn.autocommit = True |
| 98 | self.dialect = Dialect.MZ if "Materialize" in self.version() else Dialect.PG |
| 99 | |
| 100 | def close(self) -> None: |
| 101 | self.conn.close() |
Callers
nothing calls this directly
Tested by
no test coverage detected