Skip to content

Interface

interface

Package containing a repo_factory that provides an interface to data sources following the repository pattern.

Example

A XPlangGML 6.0 file can be loaded like this: 

repo = repo_factory("xplan.gml", "6.0", "xplan")
collection = repo.get_all()

repo_factory(datasource='', version=None, data_type='xplan', repo_type=None)

Factory method for Repositories.

Enables retrieving a collection of plan features from and writing to different files and DB Coretables.

Supported are:

  • GML
  • JSON-FG
  • DB (PostgreSQL, GPKG, SQLite)
  • Shapefiles

The corresponding wrapper is infered from the datasource input parameter. Currently, only the export to gml functionality of the GMLRepository is supported for INSPIRE data. Also, XTrasse is only supported for GMLRepository and DBRepository.

Parameters:

Name Type Description Default
datasource str

Name of the input source or output file.

 ''
version str | None

Version the plan data.

 None
data_type Literal['xplan', 'xtrasse', 'plu']

Specification of either xplan, xtrasse or INSPIRE plu.

 'xplan'
repo_type Literal['gml', 'jsonfg', 'shape', 'db'] | None

Allows to explicitly select a Repository.

 None

Raises:

Type Description
ValueError

raises error for unknown/unspecified datasource

Returns:

Name Type Description
BaseRepository BaseRepository

instance of repository class for manipulating a collection of plan features
Source code in xplan_tools/interface/__init__.py 
21
22
23
24
25
26
27
28
29
30
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
def repo_factory(
    datasource: str = "",
    version: str | None = None,
    data_type: Literal["xplan", "xtrasse", "plu"] = "xplan",
    repo_type: Literal["gml", "jsonfg", "shape", "db"] | None = None,
) -> "BaseRepository":
    """Factory method for Repositories.

    Enables retrieving a collection of plan features from and writing to different files and DB Coretables.

    Supported are:

    - GML
    - JSON-FG
    - DB (PostgreSQL, GPKG, SQLite)
    - Shapefiles

    The corresponding wrapper is infered from the datasource input parameter.
    Currently, only the export to gml functionality of the GMLRepository is supported for INSPIRE data.
    Also, XTrasse is only supported for GMLRepository and DBRepository.

    Args:
        datasource: Name of the input source or output file.
        version: Version the plan data.
        data_type: Specification of either xplan, xtrasse or INSPIRE plu.
        repo_type: Allows to explicitly select a Repository.

    Raises:
        ValueError: raises error for unknown/unspecified datasource

    Returns:
        BaseRepository: instance of repository class for manipulating a collection of plan features
    """
    if repo_type == "gml" or ".gml" in datasource:
        logger.debug("initializing GML repository")
        return locate("xplan_tools.interface.gml.GMLRepository")(
            datasource, version, data_type
        )
    elif repo_type == "jsonfg" or ".json" in datasource:
        logger.debug("initializing JSON-FG repository")
        return locate("xplan_tools.interface.jsonfg.JsonFGRepository")(
            datasource, version, data_type
        )
    elif repo_type == "shape" or ".shp" in datasource:
        return locate("xplan_tools.interface.shape.ShapeRepository")(
            datasource, version
        )
    elif repo_type == "db" or datasource[:4] in ["post", "gpkg", "sqli"]:
        logger.debug("initializing DB repository")
        return locate("xplan_tools.interface.db.DBRepository")(datasource, version)
    else:
        raise ValueError("Unknown datasource")

BaseRepository(datasource, version=None)

Bases: Protocol

Abstract base class for specific repositories, which re-implement supported methods for a given data source.

Initialize the Repository.

Parameters:

Name Type Description Default
datasource str

Generally a file path or URL. May also accept file-like objects.

 required
version str | None

The application schema version.

 None
Source code in xplan_tools/interface/base.py 
 9
10
11
12
13
14
15
16
17
def __init__(self, datasource: str, version: str | None = None):
    """Initialize the Repository.

    Args:
        datasource: Generally a file path or URL. May also accept file-like objects.
        version: The application schema version.
    """
    self.datasource = datasource
    self.version = version

delete(obj_id)

Delete a BaseFeature.

Source code in xplan_tools/interface/base.py 
51
52
53
def delete(self, obj_id: str) -> BaseFeature:
    """Delete a BaseFeature."""
    raise NotImplementedError("delete not implemented")

delete_plan_by_id(plan_id)

Delete a plan object with its related features.

Source code in xplan_tools/interface/base.py 
31
32
33
def delete_plan_by_id(self, plan_id: str) -> None:
    """Delete a plan object with its related features."""
    raise NotImplementedError("delete_plan_by_id not implemented")

get(obj_id)

Get a specific BaseFeature by id.

Source code in xplan_tools/interface/base.py 
19
20
21
def get(self, obj_id: str) -> BaseFeature:
    """Get a specific BaseFeature by id."""
    raise NotImplementedError("get not implemented")

get_all()

Get all BaseFeatures.

Source code in xplan_tools/interface/base.py 
23
24
25
def get_all(self) -> BaseCollection:
    """Get all BaseFeatures."""
    raise NotImplementedError("get_all not implemented")

get_plan_by_id(plan_id)

Get a plan object with its related features.

Source code in xplan_tools/interface/base.py 
27
28
29
def get_plan_by_id(self, plan_id: str) -> BaseCollection:
    """Get a plan object with its related features."""
    raise NotImplementedError("get_plan_by_id not implemented")

patch(obj_id, partial_obj)

Partially update a BaseFeature.

Source code in xplan_tools/interface/base.py 
47
48
49
def patch(self, obj_id: str, partial_obj: dict) -> BaseFeature:
    """Partially update a BaseFeature."""
    raise NotImplementedError("patch not implemented")

save(obj)

Store a BaseFeature.

Source code in xplan_tools/interface/base.py 
35
36
37
def save(self, obj: BaseFeature) -> None:
    """Store a BaseFeature."""
    raise NotImplementedError("save not implemented")

save_all(features)

Store a BaseFeature.

Source code in xplan_tools/interface/base.py 
39
40
41
def save_all(self, features: BaseCollection) -> None:
    """Store a BaseFeature."""
    raise NotImplementedError("save_all not implemented")

update(obj_id, new_obj)

Update a BaseFeature.

Source code in xplan_tools/interface/base.py 
43
44
45
def update(self, obj_id: str, new_obj: BaseFeature) -> BaseFeature:
    """Update a BaseFeature."""
    raise NotImplementedError("update not implemented")

GMLRepository(datasource='', version=None, data_type='xplan')

Bases: BaseRepository

Repository class for loading from and writing to GML files or file-like objects.

Given a plan, either xplan or INSPIRE PLU, the data is saved as GML with according namespaces and structure. Reading data from datasource and retrieving the data version is currently only supported for xplan data.

Initializes the GML Repository.

Parameters:

Name Type Description Default
datasource str | IO

A file path as a String or a file-like object.

 ''
version str | None

If no explicit version is provided it is attempted to derive the version from xplan namespace.

 None
data_type Literal['xplan', 'xtrasse', 'plu']

The GML Repository can be used for XPlanGML and INSPIRE PLU.

 'xplan'
Source code in xplan_tools/interface/gml.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    datasource: str | IO = "",
    version: str | None = None,
    data_type: Literal["xplan", "xtrasse", "plu"] = "xplan",
) -> None:
    """Initializes the GML Repository.

    Args:
        datasource: A file path as a String or a file-like object.
        version: If no explicit version is provided it is attempted to derive the version from xplan namespace.
        data_type: The GML Repository can be used for XPlanGML and INSPIRE PLU.
    """
    self.datasource = datasource
    self.data_type = data_type
    self.version = (
        self._get_version() if version is None else version
    )  # _get_version() only defined for data_type=xplan

content property

The parsed XML tree.

get_all(**kwargs)

Retrieves a Feature Collection to the datasource.

Parameters:

Name Type Description Default
**kwargs dict

Not used in this repository.

 {}
Source code in xplan_tools/interface/gml.py 
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
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
def get_all(self, **kwargs: dict) -> BaseCollection:
    """Retrieves a Feature Collection to the datasource.

    Args:
        **kwargs: Not used in this repository.
    """

    def update_related_features():
        for xlink in root.findall(".//*[@{http://www.w3.org/1999/xlink}href]"):
            href = xlink.get("{http://www.w3.org/1999/xlink}href")
            if new_id := id_mapping.get(href[1:], None):
                xlink.set("{http://www.w3.org/1999/xlink}href", f"#{new_id}")

    def validate_gml_id(feature: etree._Element):
        gml_id = feature.get("{http://www.opengis.net/gml/3.2}id")
        if not parse_uuid(gml_id):
            new_id = f"GML_{uuid4()}"
            feature.set("{http://www.opengis.net/gml/3.2}id", new_id)
            id_mapping[gml_id] = new_id
            logger.info(f"GML ID '{gml_id}' replaced with UUIDv4 '{new_id}'")

    root: etree._ElementTree = self.content

    try:
        if etree.QName(root).namespace in [
            "http://www.opengis.net/wfs/2.0",
            "http://www.opengis.net/ogcapi-features-1/1.0/sf",
        ]:
            elem = root.find(
                "./{http://www.opengis.net/gml/3.2}boundedBy/{http://www.opengis.net/gml/3.2}Envelope"
            ) or next(root.iterfind(".//*[@srsName]"))
            srs = elem.get("srsName")
        else:
            srs = root.find(
                "./{http://www.opengis.net/gml/3.2}boundedBy/{http://www.opengis.net/gml/3.2}Envelope"
            ).get("srsName")
    except (AttributeError, StopIteration, KeyError):
        raise ValueError("No SRS could be found")
    else:
        srid = parse_srs(srs)

    collection = {}
    id_mapping = {}

    for feature in root.iterfind("./*/*"):
        if etree.QName(feature).namespace == "http://www.opengis.net/gml/3.2":
            continue
        elif etree.QName(feature).namespace == "http://www.opengis.net/wfs/2.0":
            for additional_object in feature.iterfind("./*/*"):
                validate_gml_id(additional_object)
        else:
            validate_gml_id(feature)

    if id_mapping:
        update_related_features()

    for feature in root.iterfind("./*/*"):
        if etree.QName(feature).namespace == "http://www.opengis.net/gml/3.2":
            continue
        elif etree.QName(feature).namespace == "http://www.opengis.net/wfs/2.0":
            for additional_object in feature.iterfind("./*/*"):
                # set_srid_for_geom_feature(additional_object)
                model = model_factory(
                    etree.QName(additional_object).localname,
                    self.version,
                    self.data_type,
                ).model_validate(
                    additional_object,
                    context={"srid": srid, "data_type": self.data_type},
                )
                collection[model.id] = model
        else:
            # set_srid_for_geom_feature(feature)
            model = model_factory(
                etree.QName(feature).localname, self.version, self.data_type
            ).model_validate(
                feature, context={"srid": srid, "data_type": self.data_type}
            )
            collection[model.id] = model
    return BaseCollection(features=collection, srid=srid)

save_all(features, **kwargs)

Saves a Feature Collection to the datasource.

Parameters:

Name Type Description Default
features BaseCollection

A BaseCollection instance.

 required
**kwargs dict

Not used in this repository.

 {}
Source code in xplan_tools/interface/gml.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
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
def save_all(self, features: BaseCollection, **kwargs: dict) -> None:
    """Saves a Feature Collection to the datasource.

    Args:
        features: A BaseCollection instance.
        **kwargs: Not used in this repository.
    """
    if self.data_type == "xplan":
        nsmap = {
            None: f"http://www.xplanung.de/xplangml/{self.version.replace('.', '/')}",
            "gml": "http://www.opengis.net/gml/3.2",
            "xlink": "http://www.w3.org/1999/xlink",
            "xsi": "http://www.w3.org/2001/XMLSchema-instance",
        }
        root = etree.Element(
            "XPlanAuszug",
            attrib={
                "{http://www.w3.org/2001/XMLSchema-instance}schemaLocation": f"{nsmap[None]} https://repository.gdi-de.org/schemas/de.xleitstelle.xplanung/{self.version}/XPlanung-Operationen.xsd",
                "{http://www.opengis.net/gml/3.2}id": f"GML_{uuid4()}",
            },
            nsmap=nsmap,
        )
    elif self.data_type == "xtrasse":
        nsmap = {
            None: f"http://www.xtrasse.de/{self.version}",
            "gml": "http://www.opengis.net/gml/3.2",
            "xml": "http://www.w3.org/XML/1998/namespace",
            "xlink": "http://www.w3.org/1999/xlink",
            "xsi": "http://www.w3.org/2001/XMLSchema-instance",
            "sf": "http://www.opengis.net/ogcapi-features-1/1.0/sf",
        }

        root = etree.Element(
            "{http://www.opengis.net/ogcapi-features-1/1.0/sf}FeatureCollection",
            attrib={
                "{http://www.w3.org/2001/XMLSchema-instance}schemaLocation": f"{nsmap[None]} https://repository.gdi-de.org/schemas/de.xleitstelle.xtrasse/2.0/XML/XTrasse.xsd {nsmap['sf']} http://schemas.opengis.net/ogcapi/features/part1/1.0/xml/core-sf.xsd {nsmap['gml']} https://schemas.opengis.net/gml/3.2.1/gml.xsd",
                "{http://www.opengis.net/gml/3.2}id": f"GML_{uuid4()}",
            },
            nsmap=nsmap,
        )

    elif self.data_type == "plu":
        nsmap = {
            None: "http://inspire.ec.europa.eu/schemas/plu/4.0",
            "gss": "http://www.isotc211.org/2005/gss",
            "xsi": "http://www.w3.org/2001/XMLSchema-instance",
            "gco": "http://www.isotc211.org/2005/gco",
            "gml": "http://www.opengis.net/gml/3.2",
            "base": "http://inspire.ec.europa.eu/schemas/base/3.3",
            "lunom": "http://inspire.ec.europa.eu/schemas/lunom/4.0",
            "base2": "http://inspire.ec.europa.eu/schemas/base2/2.0",
            "gmd": "http://www.isotc211.org/2005/gmd",
            "xlink": "http://www.w3.org/1999/xlink",
            "wfs": "http://www.opengis.net/wfs/2.0",
        }

        root = etree.Element(
            "{http://www.opengis.net/wfs/2.0}FeatureCollection",
            attrib={
                "{http://www.w3.org/2001/XMLSchema-instance}schemaLocation": f"{nsmap[None]} https://inspire.ec.europa.eu/schemas/plu/4.0/PlannedLandUse.xsd {nsmap['wfs']} https://schemas.opengis.net/wfs/2.0/wfs.xsd {nsmap['gml']} https://schemas.opengis.net/gml/3.2.1/gml.xsd"
            },
            nsmap=nsmap,
        )

    if self.data_type != "xtrasse":
        bounds = etree.SubElement(
            root,
            "{http://www.opengis.net/gml/3.2}boundedBy"
            if self.data_type == "xplan"
            else "{http://www.opengis.net/wfs/2.0}boundedBy",
        )

    geoms = []
    feature_number = 0
    for feature in features.get_features():
        if feature:
            feature_number += 1
            if (geom_wkt := feature.get_geom_wkt()) and (
                "Plan" in feature.get_name()
            ):
                geoms.append(geom_wkt)
                srs = feature.get_geom_srid()
            etree.SubElement(
                root,
                "{http://www.opengis.net/gml/3.2}featureMember"
                if self.data_type == "xplan"
                else "{http://www.opengis.net/ogcapi-features-1/1.0/sf}featureMember"
                if self.data_type == "xtrasse"
                else "{http://www.opengis.net/wfs/2.0}member",
            ).append(
                feature.model_dump_gml(
                    self.data_type, feature_srs=kwargs.get("feature_srs", True)
                )
            )
    bbox = get_envelope(geoms)
    attrib = (
        {
            "srsName": f"http://www.opengis.net/def/crs/EPSG/0/{srs}"
        }  # TODO: Anpassung für Fälle abseits von crs?
        if self.data_type == "plu"
        else {"srsName": f"EPSG:{srs}"}
    )

    if self.data_type != "xtrasse":
        envelope = etree.SubElement(
            bounds,
            "{http://www.opengis.net/gml/3.2}Envelope",
            attrib=attrib,
        )
        etree.SubElement(
            envelope, "{http://www.opengis.net/gml/3.2}lowerCorner"
        ).text = f"{bbox[0]} {bbox[2]}"
        etree.SubElement(
            envelope, "{http://www.opengis.net/gml/3.2}upperCorner"
        ).text = f"{bbox[1]} {bbox[3]}"

    if self.data_type == "plu":
        root.set("numberMatched", str(feature_number))
        root.set("numberReturned", str(feature_number))
        root.set("timeStamp", str(datetime.datetime.now().isoformat()))

    tree = etree.ElementTree(root)
    # tree.write(
    #     self.datasource, pretty_print=True, xml_declaration=True, encoding="UTF-8"
    # )
    self._write_to_datasource(tree)

JsonFGRepository(datasource='', version=None, data_type='xplan')

Bases: BaseRepository

Repository class for loading from and writing to JSON-FG files or file-like objects.

Initializes the JSON-FG Repository.

Parameters:

Name Type Description Default
datasource str | IO

A file path as a String or a file-like object.

 ''
version str | None

If no explicit version is provided it is attempted to derive the version from the links object if its "rel" is "describedBy".

 None
Source code in xplan_tools/interface/jsonfg.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(
    self,
    datasource: str | IO = "",
    version: str | None = None,
    data_type: Literal["xplan", "xtrasse", "plu"] = "xplan",
) -> None:
    """Initializes the JSON-FG Repository.

    Args:
        datasource: A file path as a String or a file-like object.
        version: If no explicit version is provided it is attempted to derive the version from the links object if its "rel" is "describedBy".
    """
    self.datasource = datasource
    self.data_type = data_type
    self.version = self._get_version() if version is None else version

content property

The JSON data as a dict.

get_all(**kwargs)

Retrieves a Feature Collection to the datasource.

Parameters:

Name Type Description Default
**kwargs dict

Not used in this repository.

 {}
Source code in xplan_tools/interface/jsonfg.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
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
def get_all(self, **kwargs: dict) -> BaseCollection:
    """Retrieves a Feature Collection to the datasource.

    Args:
        **kwargs: Not used in this repository.
    """

    def update_related_features():
        for feature in self.content["features"]:
            model = model_factory(
                feature["featureType"], self.version, self.data_type
            )
            assoc = model.get_associations()
            for k, v in feature["properties"].items():
                if k in assoc:
                    if isinstance(v, list):
                        for i, item in enumerate(v):
                            if isinstance(item, str) and (
                                new_id := id_mapping.get(item, None)
                            ):
                                feature["properties"][k][i] = new_id
                    elif isinstance(v, str) and (new_id := id_mapping.get(v, None)):
                        feature["properties"][k] = new_id

    srid = parse_srs(self.content.get("coordRefSys", None))
    collection = {}
    id_mapping = {}

    for feature in self.content["features"]:
        feature_id = feature["id"]
        if not parse_uuid(feature_id, exact=True):
            new_id = str(uuid4())
            feature["id"] = new_id
            id_mapping[feature_id] = new_id
            logger.info(
                f"Feature ID '{feature_id}' replaced with UUIDv4 '{new_id}'"
            )

    if id_mapping:
        update_related_features()

    for feature in self.content["features"]:
        if not srid:
            srid = parse_srs(feature.get("coordRefSys", "EPSG:4326"))
        model = model_factory(
            feature["featureType"], self.version, self.data_type
        ).model_validate(
            feature, context={"srid": srid, "data_type": self.data_type}
        )
        collection[model.id] = model
    return BaseCollection(features=collection, srid=srid)

save_all(features, **kwargs)

Saves a Feature Collection to the datasource.

Parameters:

Name Type Description Default
features BaseCollection

A BaseCollection instance.

 required
**kwargs dict

Keyword arguments to pass on to model_dump_jsonfg().

 {}
Source code in xplan_tools/interface/jsonfg.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def save_all(self, features: BaseCollection, **kwargs: dict) -> None:
    """Saves a Feature Collection to the datasource.

    Args:
        features: A BaseCollection instance.
        **kwargs: Keyword arguments to pass on to [`model_dump_jsonfg()`][xplan_tools.model.base.BaseFeature.model_dump_jsonfg].
    """
    if kwargs.get("single_collection", True):
        collection = self._collection_template(srid=features.srid)
        collection["features"].extend(
            feature.model_dump_jsonfg(**kwargs)
            for feature in features.get_features()
            if feature
        )
        self._write_to_datasource(collection)
    else:
        featuretypes = {}
        for feature in features.features.values():
            featuretypes.setdefault(feature.get_name(), []).append(feature)
        for featuretype, features in featuretypes.items():
            collection = self._collection_template(
                srid=features.srid, featuretype=featuretype
            )
            collection["features"].extend(
                feature.model_dump_jsonfg(**kwargs, write_featuretype=False)
                for feature in features
            )
            self._write_to_datasource(collection)

DBRepository(datasource='', version=None, data_type='xplan')

Bases: BaseRepository

Repository class for loading from and writing to databases.

Initializes the DB Repository.

Parameters:

Name Type Description Default
datasource str

A connection string which will be transformed to a URL instance.

 ''
version str | None

If no explicit version is provided it is read from the respective table column.

 None
Source code in xplan_tools/interface/db.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(
    self,
    datasource: str = "",
    version: str | None = None,
    data_type: Literal["xplan", "xtrasse", "plu"] = "xplan",
) -> None:
    """Initializes the DB Repository.

    Args:
        datasource: A connection string which will be transformed to a URL instance.
        version: If no explicit version is provided it is read from the respective table column.
    """
    self.datasource: URL = make_url(datasource)
    self.content = None
    self.version = version
    self.dialect = self.datasource.get_dialect().name
    self.Session = sessionmaker(bind=self._engine)
    # self.session = self.Session()
    self.data_type = data_type

create_tables(srid)

Creates coretable and related/spatial tables in the database.

Parameters:

Name Type Description Default
srid int

the EPSG code for spatial data

 required
Source code in xplan_tools/interface/db.py 
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
def create_tables(self, srid: int) -> None:
    """Creates coretable and related/spatial tables in the database.

    Args:
        srid: the EPSG code for spatial data

    """
    logger.debug(f"creating tables with srid {srid}")
    tables = Base.metadata.sorted_tables
    if not self.dialect == "geopackage":
        tables.pop(1)
    try:
        tables[0].append_column(
            Column(
                "geometry", Geometry(srid=srid, spatial_index=True), nullable=True
            ),
            replace_existing=True,
        )
        if self.dialect == "sqlite":
            with self._engine.connect() as conn:
                conn.execute(text("SELECT InitSpatialMetaData('EMPTY')"))
                conn.execute(text(f"SELECT InsertEpsgSrid({srid})"))
                conn.commit()
        Base.metadata.create_all(self._engine, tables)
    except Exception as e:
        if self.dialect in ["sqlite", "geopackage"]:
            file = self._engine.url.database
            if os.path.exists(file):
                os.remove(file)
        raise e

delete_tables()

Deletes coretable and related/spatial tables from the database.

Source code in xplan_tools/interface/db.py 
246
247
248
249
def delete_tables(self) -> None:
    """Deletes coretable and related/spatial tables from the database."""
    logger.debug("deleting tables")
    Base.metadata.drop_all(self._engine)

ShapeRepository(datasource='', data_type='xplan')

Bases: BaseRepository

Repository class for collecting plans from shape files.

Given a shape file conforming to the format described here (Appendix 2), plan data is read to XPlanung classes. Only reading is supported.

Source code in xplan_tools/interface/shape.py 
30
31
32
33
34
35
36
37
def __init__(
    self,
    datasource: str = "",
    data_type: Literal["xplan"] = "xplan",
) -> None:
    self.datasource = datasource
    self.data_type = data_type
    self.version = "6.0"