nominatim-suggest/DBConnection.py at master · osm-search/nominatim-suggest · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
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
import psycopg2
from psycopg2.extras import RealDictCursor, DictCursor, register_hstore
import json
import logging


class DBConnection:
    def __init__(self, user="nominatim", password=""):
        '''
            Initializes a DB Connection

                parameters :
                    user: The username to connect to psql
                    password: password of this user
        '''
        self.connection = self.connect_to_db(user, password)

    def test_connection(self):
        '''
            Tests the connection to DB by getting dsn parameters of the connection
            and executing a test query
        '''
        self.connection.get_dsn_parameters()
        self.connection.fetch_test()

    def connect_to_db(self, user, password):
        '''
        Tries to create a connection to the Nominatim postresql DB

                parameters :
                    user: The username to connect to psql
                    password: password of this user
        '''
        logging.debug("=================================================================")
        logging.debug("Trying to create a connection")
        try:
            connection = psycopg2.connect(
                user=user,                  # Provide a user with read only permission
                                            # to avoid security issues

                password=password,          # Replace with your password while using
                host="127.0.0.1",
                port="5432",
                database="nominatim"
            )

            # Nominatim uses hstore datatype to store few fields like name.
            # register_hstore will help fetch those fields as dictionary objects.
            register_hstore(connection, globally=True, unicode=True)
            logging.debug("Success")
            return connection
        except psycopg2.errors.DatabaseError as e:
            print(e)
            exit

    def get_dsn_parameters(self):
        '''
        logging.debugs the DSN parameters of the existing connection for debugging.
        '''
        logging.debug("=================================================================")
        logging.debug("Trying to logging.debug DSN parameters: ")
        try:
            logging.debug(self.connection.connection.get_dsn_parameters())
        except psycopg2.errors.DatabaseError as e:
            print(e)
            exit

    def fetch_test(self):
        '''
        Executes a test query on the connection
        '''
        logging.debug("=================================================================")
        logging.debug("Trying to fetch from placex table: ")

        try:
            sql = "SELECT * from placex limit 1;"
            cursor = self.connection.cursor(cursor_factory=DictCursor)
            cursor.execute(sql)
            record = cursor.fetchone()
            logging.debug(sql, "\n")
            logging.debug(json.dumps(record))
            cursor.close()

        except psycopg2.errors.DataError as e:
            print(e)
            exit

    def close_connection(self):
        '''
        Closes the connection to the DB
        '''
        self.connection.close()