|
1
|
1 |
|
from typing import Any, Dict, Union |
|
2
|
|
|
|
|
3
|
1 |
|
import psycopg2 |
|
4
|
|
|
|
|
5
|
1 |
|
from pystratum_pgsql.PgSqlConnector import PgSqlConnector |
|
6
|
|
|
|
|
7
|
|
|
|
|
8
|
1 |
|
class PgSqlDefaultConnector(PgSqlConnector): |
|
9
|
|
|
""" |
|
10
|
|
|
Connects to a PostgreSQL instance using user name and password. |
|
11
|
|
|
""" |
|
12
|
|
|
|
|
13
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
|
14
|
1 |
|
def __init__(self, params: Dict[str, Union[str, int]]): |
|
15
|
|
|
""" |
|
16
|
|
|
Object constructor. |
|
17
|
|
|
|
|
18
|
|
|
:param params: The connection parameters. |
|
19
|
|
|
""" |
|
20
|
1 |
|
self._params: Dict[str, Union[str, int]] = params |
|
21
|
|
|
|
|
22
|
1 |
|
self._connection: Any = None |
|
23
|
1 |
|
""" |
|
24
|
|
|
The connection between Python and the PostgreSQL instance. |
|
25
|
|
|
""" |
|
26
|
|
|
|
|
27
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
|
28
|
1 |
|
def connect(self) -> Any: |
|
29
|
|
|
""" |
|
30
|
|
|
Connects to a PostgreSQL instance. |
|
31
|
|
|
|
|
32
|
|
|
:rtype: psycopg2.extensions.connection |
|
33
|
|
|
""" |
|
34
|
1 |
|
self._connection = psycopg2.connect(host=self._params['host'], |
|
35
|
|
|
user=self._params['user'], |
|
36
|
|
|
password=self._params['password'], |
|
37
|
|
|
database=self._params['database'], |
|
38
|
|
|
port=self._params['port']) |
|
39
|
1 |
|
cursor = self._connection.cursor() |
|
40
|
1 |
|
cursor.execute('set search_path to %s', (self._params['schema'],)) |
|
41
|
|
|
|
|
42
|
1 |
|
return self._connection |
|
43
|
|
|
|
|
44
|
|
|
# ------------------------------------------------------------------------------------------------------------------ |
|
45
|
1 |
|
def disconnect(self) -> None: |
|
46
|
|
|
""" |
|
47
|
|
|
Disconnects from a PostgreSQL instance. |
|
48
|
|
|
""" |
|
49
|
1 |
|
if self._connection: |
|
50
|
1 |
|
self._connection.close() |
|
51
|
1 |
|
self._connection = None |
|
52
|
|
|
|
|
53
|
|
|
# ---------------------------------------------------------------------------------------------------------------------- |
|
54
|
|
|
|
DatabaseStratum /
py-stratum-pgsql
