""" Module to implement Postgres services as a protocol. """ import re from MG.tools.postgis_api import External_PostGis from tools import const from tools.protocols.masterProtocol import MasterProtocol import psycopg2 class PostgreSQL(MasterProtocol): # Required data """ Class to control the PostgreSQL protocol Attributes ---------- CONNECTION_REQUIRED: dict Mapping of required connection parameters. Only is important when GMS is used with Itasker. NAME: str Protocol identifier string Parameters ---------- ip: str Url of service port: int Port of service user: str User of service connection Postgres password: str Password of service dbname: str DataBase name of service """ CONNECTION_REQUIRED = {'user': 'user', 'password': 'password', 'ip': 'ip', 'layerName': 'table_view_name', 'databaseName': 'database_name', 'schema': 'schema', 'port': 'port', 'domain': 'domain', 'subtype': 'type'} NAME = const.POSTGRESQL_KEY def __init__(self, ip, port, user, password, dbname): super().__init__() self.ip = ip self.port = port if port is not None and port != '' and int(port) > 0 \ else 5432 self.user = user self.password = password self.dbname = dbname def exists_layer(self, _): """ Service to check whether the layer exists Returns ------- success: True """ return True def exists(self, layer): """ Service to check if the PostgreSQL connection exists Parameters ---------- layer: :obj:`Layer` Returns ------- bool True if successful, False otherwise. """ ext_pg = External_PostGis(self.ip, self.port, self.user, self.password, self.dbname, layer['schema'], '') conn_exit = ext_pg.test_connection() if 'Could not connect to ' in str(conn_exit): return False return True def is_file(self, _): """ Service to check the connection results is a file Returns ------- success: True """ return True def is_directory(self, _): """ Service to check the connection results is a directory Returns ------- success:True """ return False def gdal_layer(self, layer, with_vsi=False): """ Service to get the gdal layer url Parameters ---------- layer: :obj:`Layer` with_vsi: bool To indicate if the gdal layer need vsi Returns ------- str Gdal url of the layer """ return self.gdal_url('{}.{}'.format(layer.user, layer['table_view_name']), with_vsi=with_vsi, driver_prefix=layer.driver.gdal_prefix) def gdal_url(self, source, **_): """ Service to get the gdal PostgreSQL connection url Parameters ---------- source: str Path of URL Returns ------- str Gdal url of the connection """ schema_name = source.split('.') if len(schema_name) > 2 or len(schema_name) <= 0: raise Exception('Bad schema.table_name value: {}'.format(source)) if len(schema_name) == 2: schema = schema_name[0] table = schema_name[1] else: schema = 'public' table = schema_name[0] connection_string = "PG: dbname={} host={} port={} user={} " \ "password={} {} {}". \ format(self.dbname, self.ip, self.port, self.user, self.password, "schemas={}".format(schema)*(schema != ''), "tables={}".format(table)*(table != '')) return connection_string def join(self, *_): """ Parameters ---------- _ Returns ------- pass """ return None def upload(self, *_): """ Parameters ---------- _ Returns ------- pass """ return None def download(self, *_): """ Parameters ---------- _ Returns ------- pass """ return None def format_parameters(self, _, parameters): """ Format parameters Parameters ---------- _ parameters Returns ------- """ parameters.update({ 'table_view_name': re.sub(' +', '_', parameters['table_view_name'])}) return parameters