Coverage for gws-app/gws/plugin/postgres/provider.py: 95%
101 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-10-23 20:06 +0200
1"""PostgreSQL database provider."""
3from typing import Optional
5import os
6import re
8import gws.base.database
9import gws.lib.crs
10import gws.lib.extent
11import gws.lib.net
12import gws.lib.sa as sa
13import sqlalchemy.dialects.postgresql as sa_pg
16@gws.ext.config.databaseProvider('postgres')
17class Config(gws.base.database.provider.Config):
18 """Connection to a PostgreSQL/PostGIS database."""
20 database: Optional[str]
21 """Database name."""
22 host: Optional[str]
23 """Database host."""
24 port: int = 5432
25 """Database port."""
26 username: Optional[str]
27 """User name."""
28 password: Optional[str]
29 """Password."""
30 serviceName: Optional[str]
31 """Service name from the PostgreSQL service file."""
32 options: Optional[dict]
33 """Extra libpq connection parameters."""
36@gws.ext.object.databaseProvider('postgres')
37class Object(gws.base.database.provider.Object):
38 """PostgreSQL/PostGIS database provider.
40 Connects to a PostgreSQL database and describes its tables and columns,
41 including PostGIS geometry columns, for the layers, models and finders
42 that use it.
43 """
45 def url(self):
46 return connection_url(self.config)
48 _RE_TABLE_NAME = r'''(?x)
49 ^
50 (
51 ( " (?P<a1> ([^"] | "")+ ) " )
52 |
53 (?P<a2> [^".]+ )
54 )
55 (
56 \.
57 (
58 ( " (?P<b1> ([^"] | "")+ ) " )
59 |
60 (?P<b2> [^".]+ )
61 )
62 )?
63 $
64 '''
66 _DEFAULT_SCHEMA = 'public'
68 def split_table_name(self, table_name):
69 m = re.match(self._RE_TABLE_NAME, table_name.strip())
70 if not m:
71 raise ValueError(f'invalid table name {table_name!r}')
73 d = m.groupdict()
74 s = d['a1'] or d['a2']
75 t = d['b1'] or d['b2']
76 if not t:
77 s, t = self._DEFAULT_SCHEMA, s
79 return s.replace('""', '"'), t.replace('""', '"')
81 def join_table_name(self, schema, name):
82 if not schema:
83 schema, name = self.split_table_name(name)
84 p = sa_pg.dialect().identifier_preparer
85 return p.quote(schema) + '.' + p.quote(name)
87 def table_bounds(self, table):
88 desc = self.describe(table)
89 if not desc.geometryName:
90 return
92 tab = self.table(table)
93 sql = sa.select(sa.func.ST_Extent(tab.columns.get(desc.geometryName)))
94 with self.begin() as conn:
95 box = conn.execute(sql).scalar_one()
96 extent = gws.lib.extent.from_box(box)
97 if extent:
98 return gws.Bounds(extent=extent, crs=gws.lib.crs.get(desc.geometrySrid))
100 def describe_column(self, table, column_name):
101 col = super().describe_column(table, column_name)
103 if col.nativeType == 'ARRAY':
104 sa_col = self.column(table, column_name)
105 it = getattr(sa_col.type, 'item_type', None)
106 ia = self.SA_TO_ATTR.get(type(it).__name__.upper())
107 if ia == gws.AttributeType.str:
108 col.type = gws.AttributeType.strlist
109 elif ia == gws.AttributeType.int:
110 col.type = gws.AttributeType.intlist
111 elif ia == gws.AttributeType.float:
112 col.type = gws.AttributeType.floatlist
113 else:
114 col.type = self.UNKNOWN_ARRAY_TYPE
115 return col
117 if col.nativeType == 'GEOMETRY':
118 typ, srid = self._get_geom_type_and_srid(table, column_name)
119 col.type = gws.AttributeType.geometry
120 col.geometryType = self.SA_TO_GEOM.get(typ, gws.GeometryType.geometry)
121 col.geometrySrid = srid
122 return col
124 return col
126 def _get_geom_type_and_srid(self, table, column_name):
127 """Return the geometry type and SRID of a column, from its type or from ``geometry_columns``."""
128 sa_table = self.table(table)
129 sa_col = self.column(table, column_name)
131 typ = getattr(sa_col.type, 'geometry_type', '').upper()
132 srid = getattr(sa_col.type, 'srid', 0)
134 if typ != 'GEOMETRY' and srid > 0:
135 return typ, srid
137 # not a typmod, possibly constraint-based. Query "geometry_columns"...
139 gcs = getattr(self, '_geometry_columns_cache', None)
140 if not gcs:
141 with self.begin() as conn:
142 gcs = conn.fetch_all('''
143 SELECT
144 f_table_schema,
145 f_table_name,
146 f_geometry_column,
147 type,
148 srid
149 FROM public.geometry_columns
150 ''')
151 setattr(self, '_geometry_columns_cache', gcs)
153 for gc in gcs:
154 if (
155 gc['f_table_schema'] == sa_table.schema
156 and gc['f_table_name'] == sa_table.name
157 and gc['f_geometry_column'] == sa_col.name
158 ):
159 return gc['type'], gc['srid']
161 return 'GEOMETRY', -1
164##
166def connection_url(cfg: gws.Config) -> Optional[str]:
167 """Create a connection URL from a provider configuration.
169 If ``host`` is configured, the URL is built from the host, port, database
170 and credentials. Otherwise, if ``serviceName`` is configured, the URL refers
171 to that service in the PostgreSQL service file given by ``PGSERVICEFILE``.
172 ``options`` are added as URL parameters, with ``application_name`` set to
173 ``GWS`` by default.
175 Args:
176 cfg: Provider configuration.
178 Returns:
179 The connection URL, or ``None`` if neither ``host`` nor ``serviceName`` is configured.
181 Raises:
182 ``sa.Error``: If ``serviceName`` is used and the service file does not exist.
183 """
184 # https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING
185 # https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-PARAMKEYWORDS
187 defaults = {
188 'application_name': 'GWS',
189 }
191 params = gws.u.merge(defaults, cfg.get('options'))
193 p = cfg.get('host')
194 if p:
195 return gws.lib.net.make_url(
196 scheme='postgresql',
197 username=cfg.get('username'),
198 password=cfg.get('password'),
199 hostname=p,
200 port=cfg.get('port'),
201 path=cfg.get('database') or cfg.get('dbname') or '',
202 params=params,
203 )
205 p = cfg.get('serviceName')
206 if p:
207 s = os.getenv('PGSERVICEFILE')
208 if not s or not os.path.isfile(s):
209 raise sa.Error(f'PGSERVICEFILE {s!r} not found')
211 params['service'] = p
213 return gws.lib.net.make_url(
214 scheme='postgresql',
215 hostname='',
216 path=cfg.get('database') or cfg.get('dbname') or '',
217 params=params,
218 )