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