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='', 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 inferred 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.

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

Allows to explicitly select a Repository.

None
schema

Schema name for DB repository. If not specified, the default schema is used. Only for PostgreSQL.

required
srid

The EPSG code for spatial data.

required
with_views

Whether to create geometrytype-specific views. Only for PostgreSQL.

required

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
def repo_factory(
    datasource: str = "",
    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 inferred 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.
        repo_type: Allows to explicitly select a Repository.
        schema: Schema name for DB repository. If not specified, the default schema is used. Only for PostgreSQL.
        srid: The EPSG code for spatial data.
        with_views: Whether to create geometrytype-specific views. Only for PostgreSQL.

    Raises:
        ValueError: raises error for unknown/unspecified datasource

    Returns:
        BaseRepository: instance of repository class for manipulating a collection of plan features
    """
    if isinstance(datasource, str) and datasource.startswith("postgresql://"):
        repo_type = "db"

    if not repo_type:
        try:
            gdal.PushErrorHandler("CPLQuietErrorHandler")
            gdal.UseExceptions()
            dataset = gdal.OpenEx(
                datasource, gdal.OF_VECTOR, open_options=["WRITE_GFS=NO"]
            )
            if dataset is None:
                raise RuntimeError("GDAL open failed")

            driver = dataset.GetDriver()
            driver_name = getattr(driver, "GetName", lambda: "")().lower()
            if dataset is not None:
                dataset = None
            match driver_name:
                case "postgresql" | "sqlite" | "gpkg":
                    repo_type = "db"
                case "esri shapefile":
                    repo_type = "shape"
                case "gml" | "jsonfg":
                    repo_type = driver_name
                case _:
                    raise NotImplementedError("datasource not implemented")

        except (NotImplementedError, RuntimeError) as e:
            file = Path(datasource)
            match file.suffix:
                case ".gml":
                    repo_type = "gml"
                case ".json":
                    repo_type = "jsonfg"
                case ".gpkg" | ".sqlite":
                    repo_type = "db"
                case _:
                    logger.warning(
                        f"repository type for {datasource} could not be determined: {e}"
                    )
        finally:
            gdal.PopErrorHandler()

    match repo_type:
        case "gml":
            logger.debug("initializing GML repository")
            return locate("xplan_tools.interface.gml.GMLRepository")(datasource)
        case "jsonfg":
            logger.debug("initializing JSON-FG repository")
            return locate("xplan_tools.interface.jsonfg.JsonFGRepository")(datasource)
        case "shape":
            return locate("xplan_tools.interface.shape.ShapeRepository")(datasource)
        case "db":
            logger.debug("initializing DB repository")
            return locate("xplan_tools.interface.db.DBRepository")(
                datasource,
            )
        case _:
            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
Source code in xplan_tools/interface/base.py
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.
    """
    self.datasource = datasource

delete(obj_id)

Delete a BaseFeature.

Source code in xplan_tools/interface/base.py
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
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
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
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
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
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
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
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
def update(self, obj_id: str, new_obj: BaseFeature) -> BaseFeature:
    """Update a BaseFeature."""
    raise NotImplementedError("update not implemented")

GMLRepository(datasource='')

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.

''
Source code in xplan_tools/interface/gml.py
def __init__(
    self,
    datasource: str | IO = "",
) -> None:
    """Initializes the GML Repository.

    Args:
        datasource: A file path as a String or a file-like object.
    """
    self.datasource = datasource
    self.appschema = None
    self.version = None

content property

The parsed XML tree.

get_all(always_generate_ids=False, **kwargs)

Retrieves a Feature Collection to the datasource.

Parameters:

Name Type Description Default
always_generate_ids bool

Generate new Feature IDs even if GML IDs can be parsed to UUIDs.

False
Source code in xplan_tools/interface/gml.py
def get_all(
    self, always_generate_ids: bool = False, **kwargs: dict
) -> BaseCollection:
    """Retrieves a Feature Collection to the datasource.

    Args:
        always_generate_ids: Generate new Feature IDs even if GML IDs can be parsed to UUIDs.
    """
    self.appschema = self._get_appschema()
    self.version = self._get_version()

    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")
        uuid = parse_uuid(gml_id)
        if (
            not uuid
            or (uuid and collection.get(uuid, None) == "placeholder")
            or always_generate_ids
        ):
            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}'")
        else:
            collection[uuid] = "placeholder"

    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.appschema,
                ).model_validate(
                    additional_object,
                    context={"srid": srid, "appschema": self.appschema},
                )
                collection[model.id] = model
        else:
            # set_srid_for_geom_feature(feature)
            model = model_factory(
                etree.QName(feature).localname, self.version, self.appschema
            ).model_validate(
                feature, context={"srid": srid, "appschema": self.appschema}
            )
            collection[model.id] = model
    return BaseCollection(
        features=collection,
        srid=srid,
        version=self.version,
        appschema=self.appschema,
    )

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
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.
    """
    self.appschema = features.appschema
    self.version = features.version

    if self.appschema == "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.appschema == "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.appschema == "xwp":
        nsmap = {
            None: f"http://www.xwaermeplan.de/{self.version.replace('.', '/')}",
            "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://gitlab.opencode.de/xleitstelle/xwaermeplan/spezifikation/-/raw/main/xsd/waermeplan.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.appschema == "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.appschema not in ["xtrasse", "xwp"]:
        bounds = etree.SubElement(
            root,
            (
                "{http://www.opengis.net/gml/3.2}boundedBy"
                if self.appschema == "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.appschema == "xplan"
                    else (
                        "{http://www.opengis.net/ogcapi-features-1/1.0/sf}featureMember"
                        if self.appschema in ["xtrasse", "xwp"]
                        else "{http://www.opengis.net/wfs/2.0}member"
                    )
                ),
            ).append(
                feature.model_dump_gml(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.appschema == "plu"
        else {"srsName": f"EPSG:{srs}"}
    )

    if self.appschema not in ["xtrasse", "xwp"]:
        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.appschema == "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='')

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

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

required
Source code in xplan_tools/interface/jsonfg.py
def __init__(
    self,
    datasource: str | IO = "",
) -> 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.appschema = None
    self.version = None

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
def get_all(self, **kwargs: dict) -> BaseCollection:
    """Retrieves a Feature Collection to the datasource.

    Args:
        **kwargs: Not used in this repository.
    """
    self.appschema = self._get_appschema()
    self.version = self._get_version()

    def update_related_features():
        for feature in self.content["features"]:
            model = model_factory(
                feature["featureType"], self.version, self.appschema
            )
            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.appschema
        ).model_validate(
            feature, context={"srid": srid, "appschema": self.appschema}
        )
        collection[model.id] = model
    return BaseCollection(
        features=collection,
        srid=srid,
        version=self.version,
        appschema=self.appschema,
    )

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
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].
    """
    self.appschema = features.appschema
    self.version = features.version

    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 = {}
        srid = features.srid
        for feature in features.features.values():
            featuretypes.setdefault(feature.get_name(), []).append(feature)
        for featuretype, features in featuretypes.items():
            collection = self._collection_template(
                srid=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='')

Bases: BaseRepository

Repository class for loading from and writing to databases.

Initializes the DB Repository.

During initialization, a connection is established and the existence of required tables is tested. If an alembic revision is found, automatic migration is executed for PostgreSQL DBs. For other DBs, an Exception is raised if the revision does not correspond to the current model. If no revision and tables are found, they are automatically created.

Parameters:

Name Type Description Default
datasource str

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

''
Source code in xplan_tools/interface/db.py
def __init__(
    self,
    datasource: str = "",
) -> None:
    """Initializes the DB Repository.

    During initialization, a connection is established and the existence of required tables is tested.
    If an alembic revision is found, automatic migration is executed for PostgreSQL DBs.
    For other DBs, an Exception is raised if the revision does not correspond to the current model.
    If no revision and tables are found, they are automatically created.

    Args:
        datasource: A connection string which will be transformed to a URL instance.
    """
    settings = get_settings()
    self.datasource: URL = make_url(datasource)
    self.content = None
    self.schema = settings.db_schema
    self.srid = settings.db_srid
    self.dialect = self.datasource.get_dialect().name
    self.Session = sessionmaker(bind=self._engine)

    self.alembic_cfg = config.Config()
    self.alembic_cfg.set_main_option(
        "script_location", "xplan_tools:model:migrations"
    )
    self.alembic_cfg.set_main_option(
        "sqlalchemy.url",
        datasource.replace("gpkg:", "sqlite:").replace(
            "postgresql:", "postgresql+psycopg:"
        ),
    )
    self._ensure_repo()

create_tables()

Creates coretable and related/spatial tables in the database.

Parameters:

Name Type Description Default
srid

the EPSG code for spatial data

required
Source code in xplan_tools/interface/db.py
def create_tables(self) -> None:
    """Creates coretable and related/spatial tables in the database.

    Args:
        srid: the EPSG code for spatial data
    """

    @listens_for(Base.metadata, "before_create")
    def pre_creation(_, conn, **kwargs):
        if self.dialect == "sqlite":
            conn.execute(text("SELECT InitSpatialMetaData('EMPTY')"))
            conn.execute(text("SELECT InsertEpsgSrid(:srid)"), {"srid": self.srid})

    @listens_for(Base.metadata, "after_create")
    def post_creation(_, conn, **kwargs):
        if self.dialect == "geopackage":
            conn.execute(
                text(
                    """
                    INSERT INTO gpkg_extensions (table_name, extension_name, definition, scope)
                    VALUES
                        ('gpkg_data_columns', 'gpkg_schema', 'http://www.geopackage.org/spec/#extension_schema', 'read-write'),
                        ('gpkg_data_column_constraints', 'gpkg_schema', 'http://www.geopackage.org/spec/#extension_schema', 'read-write'),
                        ('gpkgext_relations', 'related_tables', 'http://www.opengis.net/doc/IS/gpkg-rte/1.0', 'read-write'),
                        ('refs', 'related_tables', 'http://www.opengis.net/doc/IS/gpkg-rte/1.0', 'read-write')
                    """
                )
            )
            conn.execute(
                text(
                    """
                    INSERT INTO gpkgext_relations (base_table_name, base_primary_column, related_table_name, related_primary_column, relation_name, mapping_table_name)
                    VALUES
                        ('coretable', 'id', 'coretable', 'id', 'features', 'refs')
                    """
                )
            )
            conn.execute(
                text(
                    """
                    INSERT INTO gpkg_data_columns (table_name, column_name, mime_type)
                    VALUES
                        ('coretable', 'properties', 'application/json')
                    """
                )
            )

    logger.debug(f"creating tables with srid {self.srid}")
    tables = Base.metadata.sorted_tables
    if not self.dialect == "geopackage":
        tables.pop(1)
    tables[0].append_column(
        Column(
            "geometry",
            Geometry(
                srid=self.srid,
                spatial_index=True,
            ),
            nullable=True,
        ),
        replace_existing=True,
    )

    try:
        Base.metadata.create_all(self._engine, tables)
        remove(Base.metadata, "before_create", pre_creation)
        remove(Base.metadata, "after_create", post_creation)

    except Exception as e:
        if self.dialect in ["sqlite", "geopackage"]:
            file = self._engine.url.database
            Path(file).unlink(missing_ok=True)
        raise e

delete_tables()

Deletes coretable and related/spatial tables from the database.

Source code in xplan_tools/interface/db.py
def delete_tables(self) -> None:
    """Deletes coretable and related/spatial tables from the database."""
    logger.debug("deleting tables")
    if self.dialect == "postgresql":
        command.downgrade(self.alembic_cfg, "base")
    else:
        Base.metadata.drop_all(self._engine)

ShapeRepository(datasource='')

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
def __init__(
    self,
    datasource: str = "",
) -> None:
    self.datasource = datasource
    self.appschema = "xplan"
    self.version = "6.0"