API Docs
Main Functions
Parse a GeoJSON file or data struc into a normalized FeatureCollection.
Parameters:
| Name |
Type |
Description |
Default |
db
|
str | Connection
|
Existing db connection, or connection string.
|
required
|
geojson_raw
|
str | bytes | dict
|
GeoJSON file path, JSON string, dict,
or file bytes.
|
required
|
merge
|
bool
|
If any nested Polygons / MultiPolygon should be merged.
|
False
|
Returns:
| Name | Type |
Description |
FeatureCollection |
FeatureCollection
|
|
Source code in geojson_aoi/_sync/parser.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262 | def parse_aoi(
db: str | Connection, geojson_raw: str | bytes | dict, merge: bool = False
) -> FeatureCollection:
"""Parse a GeoJSON file or data struc into a normalized FeatureCollection.
Args:
db (str | Connection): Existing db connection, or connection string.
geojson_raw (str | bytes | dict): GeoJSON file path, JSON string, dict,
or file bytes.
merge (bool): If any nested Polygons / MultiPolygon should be merged.
Returns:
FeatureCollection: a FeatureCollection.
"""
# We want to maintain this list for input control.
valid_geoms = ["Polygon", "MultiPolygon", "GeometryCollection"]
# Parse different input types
if isinstance(geojson_raw, bytes):
geojson_parsed = json.loads(geojson_raw)
elif isinstance(geojson_raw, str):
if Path(geojson_raw).exists():
log.debug(f"Parsing geojson file: {geojson_raw}")
with open(geojson_raw, "rb") as geojson_file:
geojson_parsed = json.load(geojson_file)
else:
geojson_parsed = json.loads(geojson_raw)
elif isinstance(geojson_raw, dict):
geojson_parsed = geojson_raw
else:
raise ValueError("GeoJSON input must be a valid dict, str, or bytes")
# Throw error if no data
if geojson_parsed is None or geojson_parsed == {} or "type" not in geojson_parsed:
raise ValueError("Provided GeoJSON is empty")
# Throw error if wrong geometry type
if geojson_parsed["type"] not in AllowedInputTypes:
raise ValueError(f"The GeoJSON type must be one of: {AllowedInputTypes}")
# Store properties in formats that contain them.
properties = []
if (
geojson_parsed.get("type") == "Feature"
and geojson_parsed.get("geometry")
and geojson_parsed.get("geometry").get("type") in valid_geoms
):
properties.append(geojson_parsed.get("properties"))
elif geojson_parsed.get("type") == "FeatureCollection":
for feature in geojson_parsed.get("features", []):
geom = feature.get("geometry", {})
gtype = geom.get("type")
# Append a copy of the properties list for each coordinate set
# in the MultiPolygon. This ensures the split Polygons maintain
# these properties.
if gtype == "MultiPolygon":
for _coordinate in geom.get("coordinates", []):
properties.append(feature.get("properties"))
elif gtype in valid_geoms:
properties.append(feature.get("properties"))
# The same MultiPolygon handling as before.
# But applied to top-level MultiPolygons.
elif geojson_parsed.get("type") == "MultiPolygon":
# If the top-level MultiPolygon object carries properties,
# reuse them for each polygon.
top_props = geojson_parsed.get("properties")
for _coordinate in geojson_parsed.get("coordinates", []):
properties.append(top_props)
# Extract from FeatureCollection (or other types) into a
# list of Polygon geometries
geoms = strip_featcol(geojson_parsed)
# Strip away any geom type that isn't a Polygon
geoms = [geom for geom in geoms if geom and geom.get("type") == "Polygon"]
with PostGis(db, geoms, merge) as result:
# Remove any properties that PostGIS might have assigned.
for feature in result.featcol["features"]:
feature.pop("properties", None)
# Restore saved properties.
if properties:
feat_count = 0
for feature in result.featcol["features"]:
# Guard: only assign if we have a corresponding saved
# property entry.
if feat_count < len(properties):
feature["properties"] = properties[feat_count]
else:
# If for some reason counts mismatch, set to
# None rather than crashing.
feature["properties"] = None
feat_count += 1
return result.featcol
|
options:
show_source: false
heading_level: 3
Parse a GeoJSON file or data struc into a normalized FeatureCollection.
Parameters:
| Name |
Type |
Description |
Default |
db
|
str | AsyncConnection
|
Existing db connection, or connection string.
|
required
|
geojson_raw
|
str | bytes | dict
|
GeoJSON file path, JSON string, dict,
or file bytes.
|
required
|
merge
|
bool
|
If any nested Polygons / MultiPolygon should be merged.
|
False
|
Returns:
| Name | Type |
Description |
FeatureCollection |
FeatureCollection
|
|
Source code in geojson_aoi/_async/parser.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262 | async def parse_aoi_async(
db: str | AsyncConnection, geojson_raw: str | bytes | dict, merge: bool = False
) -> FeatureCollection:
"""Parse a GeoJSON file or data struc into a normalized FeatureCollection.
Args:
db (str | AsyncConnection): Existing db connection, or connection string.
geojson_raw (str | bytes | dict): GeoJSON file path, JSON string, dict,
or file bytes.
merge (bool): If any nested Polygons / MultiPolygon should be merged.
Returns:
FeatureCollection: a FeatureCollection.
"""
# We want to maintain this list for input control.
valid_geoms = ["Polygon", "MultiPolygon", "GeometryCollection"]
# Parse different input types
if isinstance(geojson_raw, bytes):
geojson_parsed = json.loads(geojson_raw)
elif isinstance(geojson_raw, str):
if Path(geojson_raw).exists():
log.debug(f"Parsing geojson file: {geojson_raw}")
with open(geojson_raw, "rb") as geojson_file:
geojson_parsed = json.load(geojson_file)
else:
geojson_parsed = json.loads(geojson_raw)
elif isinstance(geojson_raw, dict):
geojson_parsed = geojson_raw
else:
raise ValueError("GeoJSON input must be a valid dict, str, or bytes")
# Throw error if no data
if geojson_parsed is None or geojson_parsed == {} or "type" not in geojson_parsed:
raise ValueError("Provided GeoJSON is empty")
# Throw error if wrong geometry type
if geojson_parsed["type"] not in AllowedInputTypes:
raise ValueError(f"The GeoJSON type must be one of: {AllowedInputTypes}")
# Store properties in formats that contain them.
properties = []
if (
geojson_parsed.get("type") == "Feature"
and geojson_parsed.get("geometry")
and geojson_parsed.get("geometry").get("type") in valid_geoms
):
properties.append(geojson_parsed.get("properties"))
elif geojson_parsed.get("type") == "FeatureCollection":
for feature in geojson_parsed.get("features", []):
geom = feature.get("geometry", {})
gtype = geom.get("type")
# Append a copy of the properties list for each coordinate set
# in the MultiPolygon. This ensures the split Polygons maintain
# these properties.
if gtype == "MultiPolygon":
for _coordinate in geom.get("coordinates", []):
properties.append(feature.get("properties"))
elif gtype in valid_geoms:
properties.append(feature.get("properties"))
# The same MultiPolygon handling as before.
# But applied to top-level MultiPolygons.
elif geojson_parsed.get("type") == "MultiPolygon":
# If the top-level MultiPolygon object carries properties,
# reuse them for each polygon.
top_props = geojson_parsed.get("properties")
for _coordinate in geojson_parsed.get("coordinates", []):
properties.append(top_props)
# Extract from FeatureCollection (or other types) into a
# list of Polygon geometries
geoms = strip_featcol(geojson_parsed)
# Strip away any geom type that isn't a Polygon
geoms = [geom for geom in geoms if geom and geom.get("type") == "Polygon"]
async with AsyncPostGis(db, geoms, merge) as result:
# Remove any properties that AsyncPostGIS might have assigned.
for feature in result.featcol["features"]:
feature.pop("properties", None)
# Restore saved properties.
if properties:
feat_count = 0
for feature in result.featcol["features"]:
# Guard: only assign if we have a corresponding saved
# property entry.
if feat_count < len(properties):
feature["properties"] = properties[feat_count]
else:
# If for some reason counts mismatch, set to
# None rather than crashing.
feature["properties"] = None
feat_count += 1
return result.featcol
|
options:
show_source: false
heading_level: 3
Helpers
Database config for Postgres.
Allows overriding values via constructor parameters, fallback to env vars.
Attributes:
| Name |
Type |
Description |
dbname |
str
|
|
user |
str
|
|
password |
str
|
|
host |
str
|
|
port |
int
|
|
Source code in geojson_aoi/dbconfig.py
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 | @dataclass
class DbConfig:
"""Database config for Postgres.
Allows overriding values via constructor parameters, fallback to env vars.
Attributes:
dbname (str): Database name.
user (str): Database user.
password (str): Database password.
host (str): Database host.
port (int): Database port.
"""
dbname: Optional[str] = None
user: Optional[str] = None
password: Optional[str] = None
host: Optional[str] = None
port: Optional[int] = None
def __post_init__(self):
"""Ensures env variables are read at runtime, not at class definition."""
self.dbname = self.dbname or os.getenv("GEOJSON_AOI_DB_NAME")
self.user = self.user or os.getenv("GEOJSON_AOI_DB_USER")
self.password = self.password or os.getenv("GEOJSON_AOI_DB_PASSWORD")
self.host = self.host or os.getenv("GEOJSON_AOI_DB_HOST", "db")
self.port = self.port or int(os.getenv("GEOJSON_AOI_DB_PORT", "5432"))
# Raise error if any required field is missing
missing_fields = [
field
for field in ["dbname", "user", "password"]
if not getattr(self, field)
]
if missing_fields:
raise ValueError(
f"Missing required database config fields: {', '.join(missing_fields)}"
)
def get_connection_string(self) -> str:
"""Connection string that psycopg accepts."""
return (
f"dbname={self.dbname} user={self.user} password={self.password} "
f"host={self.host} port={self.port}"
)
|
__post_init__()
Ensures env variables are read at runtime, not at class definition.
Source code in geojson_aoi/dbconfig.py
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69 | def __post_init__(self):
"""Ensures env variables are read at runtime, not at class definition."""
self.dbname = self.dbname or os.getenv("GEOJSON_AOI_DB_NAME")
self.user = self.user or os.getenv("GEOJSON_AOI_DB_USER")
self.password = self.password or os.getenv("GEOJSON_AOI_DB_PASSWORD")
self.host = self.host or os.getenv("GEOJSON_AOI_DB_HOST", "db")
self.port = self.port or int(os.getenv("GEOJSON_AOI_DB_PORT", "5432"))
# Raise error if any required field is missing
missing_fields = [
field
for field in ["dbname", "user", "password"]
if not getattr(self, field)
]
if missing_fields:
raise ValueError(
f"Missing required database config fields: {', '.join(missing_fields)}"
)
|
get_connection_string()
Connection string that psycopg accepts.
Source code in geojson_aoi/dbconfig.py
| def get_connection_string(self) -> str:
"""Connection string that psycopg accepts."""
return (
f"dbname={self.dbname} user={self.user} password={self.password} "
f"host={self.host} port={self.port}"
)
|
options:
show_source: false
heading_level: 3