LEFT | RIGHT |
(no file at all) | |
| 1 # This file is part of trytond_gis. The COPYRIGHT file at the top level of |
| 2 # this repository contains the full copyright notices and license terms. |
| 3 import binascii |
| 4 |
| 5 from geomet import wkb |
| 6 from psycopg2.extensions import register_adapter, new_type, register_type, AsIs |
| 7 |
| 8 from trytond.backend.postgresql.database import Database as PGDatabase |
| 9 from trytond.backend.postgresql.database import ( |
| 10 DatabaseIntegrityError, DatabaseOperationalError) |
| 11 |
| 12 from trytond_gis import _GeoJSON |
| 13 |
| 14 __all__ = ['Database', 'DatabaseIntegrityError', 'DatabaseOperationalError'] |
| 15 |
| 16 |
| 17 def ewkb2geojson(value, cursor): |
| 18 if value is None: |
| 19 return None |
| 20 return wkb.loads(binascii.a2b_hex(value)) |
| 21 |
| 22 |
| 23 class Database(PGDatabase): |
| 24 |
| 25 _GIS_OIDS = None |
| 26 |
| 27 @classmethod |
| 28 def create(cls, connection, database_name): |
| 29 super(Database, cls).create(connection, database_name) |
| 30 |
| 31 database = cls(database_name) |
| 32 with database.get_connection() as db_connection: |
| 33 cursor = db_connection.cursor() |
| 34 cursor.execute("CREATE EXTENSION postgis") |
| 35 |
| 36 def get_connection(self, autocommit=False, readonly=False): |
| 37 conn = super(Database, self).get_connection(autocommit, readonly) |
| 38 |
| 39 if self._GIS_OIDS is None: |
| 40 cursor = conn.cursor() |
| 41 cursor.execute( |
| 42 "SELECT 1 FROM pg_extension WHERE extname='postgis'") |
| 43 if cursor.fetchone(): |
| 44 cursor.execute('SELECT NULL::geometry, NULL::geography') |
| 45 geometry_oid = cursor.description[0][1] |
| 46 geography_oid = cursor.description[1][1] |
| 47 self._GIS_OIDS = { |
| 48 'geometry': geometry_oid, |
| 49 'geography': geography_oid, |
| 50 } |
| 51 |
| 52 GEOMETRY = new_type((geometry_oid,), 'GEOMETRY', ewkb2geojson) |
| 53 register_type(GEOMETRY) |
| 54 GEOGRAPHY = new_type( |
| 55 (geography_oid,), 'GEOGRAPHY', ewkb2geojson) |
| 56 register_type(GEOGRAPHY) |
| 57 |
| 58 return conn |
| 59 |
| 60 register_adapter(_GeoJSON, |
| 61 lambda value: AsIs("'%s'" % binascii.b2a_hex(wkb.dumps(value)).upper())) |
LEFT | RIGHT |