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.

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', 'plu']

Specification of either xplan 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
def repo_factory(
    datasource: str = "",
    version: str | None = None,
    data_type: Literal["xplan", "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.

    Args:
        datasource: Name of the input source or output file.
        version: Version the plan data.
        data_type: Specification of either xplan 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
        )
    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
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
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='', 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', 'plu']

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

'xplan'
Source code in xplan_tools/interface/gml.py
def __init__(
    self,
    datasource: str | IO = "",
    version: str | None = None,
    data_type: Literal["xplan", "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
def get_all(self, **kwargs: dict) -> BaseCollection:
    """Retrieves a Feature Collection to the datasource.

    Args:
        **kwargs: Not used in this repository.
    """
    root: etree._ElementTree = etree.parse(self.datasource).getroot()
    # return [BaseFeature.get_class(etree.QName(feature).localname).from_etree(feature) for feature in root.xpath('./*/*[namespace-uri() != "http://www.opengis.net/gml/3.2"]')]
    collection = {}
    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("./*/*"):
                model = model_factory(
                    etree.QName(additional_object).localname, self.version
                ).model_validate(additional_object)
                collection[model.id] = model
        else:
            model = model_factory(
                etree.QName(feature).localname, self.version
            ).model_validate(feature)
            collection[model.id] = model
    return BaseCollection(root=collection)

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.
    """
    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 == "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",
            # "tn": "http://inspire.ec.europa.eu/schemas/tn/4.0",
            # "ad": "http://inspire.ec.europa.eu/schemas/ad/4.0",
            "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",
            # "gmlcov": "http://www.opengis.net/gmlcov/1.0",
            # "bu-base": "http://inspire.ec.europa.eu/schemas/bu-base/4.0",
            "lunom": "http://inspire.ec.europa.eu/schemas/lunom/4.0",
            "base2": "http://inspire.ec.europa.eu/schemas/base2/2.0",
            # "gsr": "http://www.isotc211.org/2005/gsr",
            # "au": "http://inspire.ec.europa.eu/schemas/au/4.0",
            # "gts": "http://www.isotc211.org/2005/gts",
            # "cp": "http://inspire.ec.europa.eu/schemas/cp/4.0",
            # "gn": "http://inspire.ec.europa.eu/schemas/gn/4.0",
            "gmd": "http://www.isotc211.org/2005/gmd",
            # "net": "http://inspire.ec.europa.eu/schemas/net/4.0",
            # "sc": "http://www.interactive-instruments.de/ShapeChange/AppInfo",
            # "swe": "http://www.opengis.net/swe/2.0",
            "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,
        )

    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/wfs/2.0}member",
            ).append(feature.model_dump_gml(self.data_type))
    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}"}
    )
    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)

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

    Args:
        **kwargs: Not used in this repository.
    """
    collection = {}
    for feature in self.content["features"]:
        model = model_factory(feature["featureType"], self.version).model_validate(
            feature
        )
        collection[model.id] = model
    return BaseCollection(root=collection)

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].
    """
    if kwargs.get("single_collection", True):
        collection = self._collection_template()
        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.root.values():
            featuretypes.setdefault(feature.get_name(), []).append(feature)
        for featuretype, features in featuretypes.items():
            collection = self._collection_template(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)

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

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
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)
        with self._engine.connect() as conn:
            match self.dialect:
                case "geopackage" | "sqlite":
                    conn.execute(
                        text("""
                            CREATE TRIGGER auto_increment_coretable
                                AFTER INSERT ON coretable
                                WHEN new.sort_key IS NULL
                                BEGIN
                                    UPDATE coretable
                                    SET sort_key = (SELECT IFNULL(MAX(sort_key), 0) + 1 FROM coretable)
                                    WHERE id = new.id;
                                END;
                            """)
                    )
                    conn.execute(
                        text("""
                            CREATE TRIGGER auto_increment_refs
                                AFTER INSERT ON refs
                                WHEN new.sort_key IS NULL
                                BEGIN
                                    UPDATE refs
                                    SET sort_key = (SELECT IFNULL(MAX(sort_key), 0) + 1 FROM refs)
                                    WHERE base_id = new.base_id AND related_id = new.related_id;
                                END;
                            """)
                    )
                    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')
                                """)
                        )
                case "postgresql":
                    conn.execute(
                        text("""
                            CREATE OR REPLACE FUNCTION increment_coretable()
                            RETURNS TRIGGER
                            LANGUAGE PLPGSQL
                            AS $$
                            BEGIN
                                UPDATE coretable
                                SET sort_key = (SELECT COALESCE(MAX(sort_key), 0) + 1 FROM coretable)
                                WHERE id = new.id;
                                RETURN NULL;
                            END;
                            $$
                            """)
                    )
                    conn.execute(
                        text("""
                            CREATE OR REPLACE FUNCTION increment_refs()
                            RETURNS TRIGGER
                            LANGUAGE PLPGSQL
                            AS $$
                            BEGIN
                                UPDATE refs
                                SET sort_key = (SELECT COALESCE(MAX(sort_key), 0) + 1 FROM refs)
                                WHERE base_id = new.base_id AND related_id = new.related_id;
                                RETURN NULL;
                            END;
                            $$
                            """)
                    )
                    conn.execute(
                        text("""
                            CREATE OR REPLACE TRIGGER auto_increment_coretable
                                AFTER INSERT
                                ON coretable
                                FOR EACH ROW
                                EXECUTE PROCEDURE increment_coretable();
                            """)
                    )
                    conn.execute(
                        text("""
                            CREATE OR REPLACE TRIGGER auto_increment_refs
                                AFTER INSERT
                                ON refs
                                FOR EACH ROW
                                EXECUTE PROCEDURE increment_refs();
                            """)
                    )
            conn.commit()
    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
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
def __init__(
    self,
    datasource: str = "",
    data_type: Literal["xplan"] = "xplan",
) -> None:
    self.datasource = datasource
    self.data_type = data_type
    self.version = "6.0"