Interface

`interface`

Package providing an interface to data sources following the repository pattern.

Example

A XPlangGML 6.0 file can be loaded like this:

collection = Datasource("xplan.gml").repo().get_all()

Datasource is the value object every repository stores: it classifies a datasource, retrieves a remote one and identifies its format, and Datasource.repo builds the matching repository. It can also be used on its own, for instance to read a WFS response:

repo = Datasource(wfs_url, allow_remote=True).repo()

`Datasource(source, *, allow_remote=None, format=None, for_write=False)`

Where a datasource is, what it holds, and which repository reads it.

The value object every repository stores. Construction classifies and validates but never reads: a path is opened, and a remote URL retrieved, only when an attribute first needs the bytes.

Usage example

# read a local file
collection = Datasource("plan.gml").repo().get_all()
# read a WFS response, which carries no usable file extension
ds = Datasource(wfs_url, allow_remote=True)
ds.format         # -> "gml", identified from the retrieved content
ds.repo_class     # -> <class GMLRepository>
collection = ds.repo().get_all()
# write to a GeoPackage; a database is always addressed by connection URL
Datasource("gpkg:///out.gpkg").repo().save_all(collection)
# name the format instead of identifying it from the content
Datasource("out.dat", format="gml").repo().save_all(collection)

Attributes:

Name	Type	Description
`raw`		The datasource exactly as it was handed over.
`kind`	`Kind`	Where the datasource's bytes are.
`allow_remote`		Whether retrieval was permitted; `None` defers to the `XMAS_DS_ALLOW_REMOTE` setting.
`for_write`		Whether this is a write target, and so classified by name.

Classifies and validates a datasource, without reading it.

Parameters:

Name	Type	Description	Default
`source`	`Any`	A file path as a string or `os.PathLike`, a connection URL, an `http(s)` URL, a file-like object, or an existing `Datasource`, whose classification is reused - so handing a resolved datasource to a repository does not resolve it twice.	required
`allow_remote`	`bool \| None`	Whether an `http(s)` datasource may be retrieved; defaults to the `XMAS_DS_ALLOW_REMOTE` setting. Pass `False` for an output target, which is never fetched.	`None`
`format`	`RepoType \| None`	The repository type, when it is known. Skips identifying it.	`None`
`for_write`	`bool`	Whether this datasource is a write target, whose `format` is then taken from its name rather than from its content.	`False`

Raises:

Type	Description
`UnsupportedDatasourceError`	The datasource names a location this package refuses to open, or is remote while remote access is disabled.

Source code in xplan_tools/interface/datasource.py

def __init__(
    self,
    source: Any,
    *,
    allow_remote: bool | None = None,
    format: RepoType | None = None,
    for_write: bool = False,
) -> None:
    """Classifies and validates a datasource, without reading it.

    Args:
        source: A file path as a string or `os.PathLike`, a connection URL, an
            ``http(s)`` URL, a file-like object, or an existing `Datasource`, whose
            classification is reused - so handing a resolved datasource to a
            repository does not resolve it twice.
        allow_remote: Whether an ``http(s)`` datasource may be retrieved; defaults to
            the ``XMAS_DS_ALLOW_REMOTE`` setting. Pass `False` for an output target,
            which is never fetched.
        format: The repository type, when it is known. Skips identifying it.
        for_write: Whether this datasource is a write target, whose
            [`format`][xplan_tools.interface.datasource.Datasource] is then taken
            from its name rather than from its content.

    Raises:
        UnsupportedDatasourceError: The datasource names a location this package
            refuses to open, or is remote while remote access is disabled.
    """
    if isinstance(source, Datasource):
        # already classified; carry the caches over so nothing is fetched twice
        self.__dict__.update(source.__dict__)
        if allow_remote is not None:
            self.allow_remote = allow_remote
        if for_write and not self.for_write:
            # promoted to a write target after the fact: a format identified from
            # the content that is about to be replaced no longer applies
            self.__dict__.pop("format", None)
            self.for_write = True
        if format is not None:
            self.format = format
        # the caches came over too, so a `buffer` fetched under the policy this call
        # just revoked would otherwise be handed back as if it were still allowed
        self._guard_remote(source)
        return
    if isinstance(source, os.PathLike):
        # so that Datasource(other.uri) round-trips for a local file
        source = os.fspath(source)
    self.raw = source
    self.allow_remote = allow_remote
    self.for_write = for_write
    if format is not None:
        # seeds the cached_property below; writing None would shadow it for good
        self.format = format
    self.kind: Kind = _kind_of(source)
    self._guard_remote(source)

`buffer` `cached` `property`

The datasource's bytes, when they are in memory rather than on disk.

Retrieves a remote datasource on first access. A path is left for the repository to open, so a large file is never read into memory here. A stream that cannot be rewound is drained into a BytesIO once, so that identifying the format does not consume it and leave every later reader with nothing.

Raises:

Type	Description
`RemoteFetchError`	A remote datasource could not be retrieved.

`format` `cached` `property`

Which repository reads this datasource, identified from its content.

A write target (for_write) is identified from its name instead: whatever is at the path now is about to be replaced, so it says nothing about what is being written. Reading it would refuse to overwrite a file holding a document this package does not read.

Raises:

Type	Description
`ForbiddenDoctypeError`	The datasource is an XML document declaring a DTD.
`UnsupportedRootElementError`	The datasource is XML, but opens with an element this package does not read.
`RemoteFetchError`	A remote datasource could not be retrieved.

`is_remote` `property`

Whether the datasource is an http(s) URL.

`repo_class` `property`

The repository class that reads this datasource.

Imported on demand, one module at a time: those modules import this one in turn, so the import cannot be made at module scope. One at a time rather than all four because DBRepository drags in Alembic, which a GML file has no use for.

Raises:

Type	Description
`DatasourceError`	The datasource could not be identified.

`root_info` `cached` `property`

The root element of an XML datasource, read without parsing the rest of it.

Seeded by format when the content sniff already read it, so a GML datasource is preflighted once however it was reached.

Raises:

Type	Description
`ForbiddenDoctypeError`	The datasource is an XML document declaring a DTD.
`UnsupportedRootElementError`	The datasource opens with an element this package does not read.
`XMLParseError`	The datasource does not hold well-formed XML.

`source` `property`

What a file-based repository reads: the buffer if there is one, else the path.

DBRepository reads uri instead.

`uri` `cached` `property`

The datasource as a typed location, or None for a file-like object.

A sqlalchemy.URL masks its password, which is why this rather than raw is what gets logged. An http(s) URL is split by the standard library rather than validated by a URL model: classification has already accepted the scheme, and a stricter parse here would raise an error no caller of this package expects.

Raises:

Type	Description
`DatasourceError`	The connection URL could not be parsed. The URL itself is left out of the message, since a password cannot be masked in a string that did not parse.

`open()`

Yields the datasource as a binary stream positioned at the start.

A path is opened and closed here; a buffer is rewound and put back where it was found, so a caller's stream comes back as it was handed over.

Raises:

Type	Description
`DatasourceError`	The datasource is a database connection, which holds no stream of bytes.
`RemoteFetchError`	A remote datasource could not be retrieved.
`OSError`	The path could not be opened.

Yields:

Type	Description
`IO[bytes]`	A binary stream over the datasource.

Source code in xplan_tools/interface/datasource.py

@contextmanager
def open(self) -> Iterator[IO[bytes]]:
    """Yields the datasource as a binary stream positioned at the start.

    A path is opened and closed here; a buffer is rewound and put back where it was
    found, so a caller's stream comes back as it was handed over.

    Raises:
        DatasourceError: The datasource is a database connection, which holds no
            stream of bytes.
        RemoteFetchError: A remote datasource could not be retrieved.
        OSError: The path could not be opened.

    Yields:
        A binary stream over the datasource.
    """
    if self.kind == "db":
        raise DatasourceError(
            f"{self!r} is a database connection, not a byte stream"
        )
    with as_binary_stream(self.source) as stream:
        yield stream

`read_bytes()`

Returns the whole datasource, retrieving a remote one on the way.

Raises:

Type	Description
`DatasourceError`	The datasource is a database connection.
`RemoteFetchError`	A remote datasource could not be retrieved.
`OSError`	The path could not be opened.

Returns:

Type	Description
`bytes`	The datasource's bytes.

Source code in xplan_tools/interface/datasource.py

def read_bytes(self) -> bytes:
    """Returns the whole datasource, retrieving a remote one on the way.

    Raises:
        DatasourceError: The datasource is a database connection.
        RemoteFetchError: A remote datasource could not be retrieved.
        OSError: The path could not be opened.

    Returns:
        The datasource's bytes.
    """
    with self.open() as stream:
        return stream.read()

`repo()`

Builds the repository that reads this datasource.

Raises:

Type	Description
`DatasourceError`	The datasource could not be identified.

Returns:

Type	Description
`BaseRepository`	An instance of the matching repository class.

Source code in xplan_tools/interface/datasource.py

def repo(self) -> BaseRepository:
    """Builds the repository that reads this datasource.

    Raises:
        DatasourceError: The datasource could not be identified.

    Returns:
        An instance of the matching repository class.
    """
    repo_class = self.repo_class
    logger.debug(f"initializing {repo_class.__name__}")
    return repo_class(self)

`write_bytes(data)`

Writes data to the datasource, encoding for a text buffer.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	The serialized document.	required

Raises:

Type	Description
`DatasourceError`	The datasource is not something this package writes to - a database connection, which its repository writes through SQLAlchemy, or a remote URL, which is never written back.
`OSError`	The path could not be written.

Returns:

Type	Description
`int`	The number of bytes, or characters for a text buffer, written.

Source code in xplan_tools/interface/datasource.py

def write_bytes(self, data: bytes) -> int:
    """Writes `data` to the datasource, encoding for a text buffer.

    Args:
        data: The serialized document.

    Raises:
        DatasourceError: The datasource is not something this package writes to - a
            database connection, which its repository writes through SQLAlchemy, or
            a remote URL, which is never written back.
        OSError: The path could not be written.

    Returns:
        The number of bytes, or characters for a text buffer, written.
    """
    match self.kind:
        case "path":
            written = Path(self.raw).write_bytes(data)
        case "buffer":
            written = self._replace_buffer(
                data.decode("utf-8")
                if isinstance(self.raw, io.TextIOBase)
                else data
            )
        case _:
            raise DatasourceError(f"{self!r} cannot be written to")
    # Whatever was read out of the datasource describes what used to be there. A text
    # buffer is the one kind where that is not self-correcting: `buffer` holds an
    # encoded snapshot rather than `raw` itself, so without this a read after a write
    # returns the document from before it. `format` is deliberately kept - it is the
    # routing decision the caller made, not an observation of the content.
    self.__dict__.pop("buffer", None)
    self.__dict__.pop("root_info", None)
    return written

`fetch_remote(url, *, allow_remote=None, timeout=None, max_bytes=None)`

Retrieves a remote datasource into a buffer.

The buffer can be handed straight to a repository. Retrieval happens here, rather than inside lxml, so that the parser needs no network access of its own.

Parameters:

Name	Type	Description	Default
`url`	`str`	An `http` or `https` URL.	required
`allow_remote`	`bool \| None`	Whether remote retrieval is permitted; defaults to the `XMAS_DS_ALLOW_REMOTE` setting.	`None`
`timeout`	`float \| None`	Seconds to wait; defaults to the `XMAS_DS_REMOTE_TIMEOUT` setting.	`None`
`max_bytes`	`int \| None`	Largest response body accepted; defaults to the `XMAS_DS_REMOTE_MAX_BYTES` setting.	`None`

Raises:

Type	Description
`UnsupportedDatasourceError`	Remote retrieval is disabled, or the URL does not name an `http(s)` resource.
`RemoteFetchError`	The request failed, or the response exceeded `max_bytes`.

Returns:

Type	Description
`BytesIO`	The response body, positioned at the start.

Source code in xplan_tools/interface/datasource.py

def fetch_remote(
    url: str,
    *,
    allow_remote: bool | None = None,
    timeout: float | None = None,
    max_bytes: int | None = None,
) -> io.BytesIO:
    """Retrieves a remote datasource into a buffer.

    The buffer can be handed straight to a repository. Retrieval happens here, rather
    than inside `lxml`, so that the parser needs no network access of its own.

    Args:
        url: An ``http`` or ``https`` URL.
        allow_remote: Whether remote retrieval is permitted; defaults to the
            ``XMAS_DS_ALLOW_REMOTE`` setting.
        timeout: Seconds to wait; defaults to the ``XMAS_DS_REMOTE_TIMEOUT`` setting.
        max_bytes: Largest response body accepted; defaults to the
            ``XMAS_DS_REMOTE_MAX_BYTES`` setting.

    Raises:
        UnsupportedDatasourceError: Remote retrieval is disabled, or the URL does not
            name an ``http(s)`` resource.
        RemoteFetchError: The request failed, or the response exceeded `max_bytes`.

    Returns:
        The response body, positioned at the start.
    """
    import httpx2  # deferred so this module does not depend on it being importable

    settings = get_settings()
    if allow_remote is None:
        allow_remote = settings.ds_allow_remote
    if timeout is None:
        timeout = settings.ds_remote_timeout
    if max_bytes is None:
        max_bytes = settings.ds_remote_max_bytes

    if urlsplit(url).scheme.lower() not in _HTTP_SCHEMES:
        raise UnsupportedDatasourceError(
            f"only http(s) datasources can be retrieved, got {url!r}"
        )
    if not allow_remote:
        raise UnsupportedDatasourceError(
            f"remote datasources are disabled, refusing to retrieve {url!r}; set "
            "XMAS_DS_ALLOW_REMOTE=1 or pass --allow-remote to enable them"
        )

    buffer = io.BytesIO()
    size = 0
    try:
        with httpx2.Client(
            timeout=timeout, follow_redirects=True, max_redirects=5
        ) as client:
            with client.stream("GET", url) as response:
                response.raise_for_status()
                for chunk in response.iter_bytes():
                    size += len(chunk)
                    if size > max_bytes:
                        raise RemoteFetchError(
                            f"{url!r} exceeds the {max_bytes} byte limit; raise "
                            "XMAS_DS_REMOTE_MAX_BYTES to allow it"
                        )
                    buffer.write(chunk)
                final_url = str(response.url)
    except httpx2.HTTPStatusError as e:
        raise RemoteFetchError(f"{url!r} returned HTTP {e.response.status_code}") from e
    except httpx2.HTTPError as e:
        raise RemoteFetchError(f"could not retrieve {url!r}: {e}") from e

    logger.info(f"retrieved {size} bytes from {final_url}")
    buffer.seek(0)
    return buffer

`repo_factory(datasource='', repo_type=None, allow_remote=None)`

Factory method for Repositories.

Deprecated: use Datasource instead, which this delegates to: Datasource(datasource, allow_remote=...).repo().

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`
`allow_remote`	`bool \| None`	Whether an `http(s)` datasource may be retrieved; defaults to the `XMAS_DS_ALLOW_REMOTE` setting. Pass `False` for an output target, which is never fetched.	`None`

Raises:

Type	Description
`DatasourceError`	raises error for unknown/unspecified datasource
`UnsupportedDatasourceError`	raises error for a datasource this package refuses to open, or a remote one while remote access is disabled
`RemoteFetchError`	raises error when a remote datasource could not be retrieved

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

@deprecated("repo_factory() is deprecated; use Datasource(...).repo() instead")
def repo_factory(
    datasource: str = "",
    repo_type: Literal["gml", "jsonfg", "shape", "db"] | None = None,
    allow_remote: bool | None = None,
) -> "BaseRepository":
    """Factory method for Repositories.

    Deprecated: use [`Datasource`][xplan_tools.interface.datasource.Datasource] instead,
    which this delegates to: `Datasource(datasource, allow_remote=...).repo()`.

    Args:
        datasource: Name of the input source or output file.
        repo_type: Allows to explicitly select a Repository.
        allow_remote: Whether an `http(s)` datasource may be retrieved; defaults to the
            `XMAS_DS_ALLOW_REMOTE` setting. Pass `False` for an output target, which is
            never fetched.

    Raises:
        DatasourceError: raises error for unknown/unspecified datasource
        UnsupportedDatasourceError: raises error for a datasource this package refuses to
            open, or a remote one while remote access is disabled
        RemoteFetchError: raises error when a remote datasource could not be retrieved

    Returns:
        BaseRepository: instance of repository class for manipulating a collection of plan features
    """
    return Datasource(datasource, allow_remote=allow_remote, format=repo_type).repo()

`sniff_datasource(source, *, suffix='')`

Identifies which repository a datasource should be read with.

Content decides: an XML document by its root element, a GeoPackage or shapefile by its magic number, JSON-FG by its first byte. So a URL with no usable file extension - a WFS GetFeature request, say - is identified from what it served. The extension is the fallback, and is all an output target that does not exist yet can offer.

A shapefile is the .shp itself, never the directory holding it: GDAL exposes each .shp in a directory as a layer and ShapeRepository reads only the first one, so naming the file is what makes the choice deterministic.

Parameters:

Name	Type	Description	Default
`source`	`str \| BytesIO`	A file path, or a buffer holding the datasource.	required
`suffix`	`str`	File extension to fall back on, e.g. `".gml"`. Ignored when `source` is a path, which carries its own.	`''`

Raises:

Type	Description
`ForbiddenDoctypeError`	The datasource is an XML document declaring a DTD.
`UnsupportedRootElementError`	The datasource is XML, but opens with an element this package does not read.

Returns:

Type	Description
`RepoType \| None`	The repository type, or `None` when the datasource could not be identified.

Source code in xplan_tools/interface/datasource.py

def sniff_datasource(source: str | io.BytesIO, *, suffix: str = "") -> RepoType | None:
    """Identifies which repository a datasource should be read with.

    Content decides: an XML document by its root element, a GeoPackage or shapefile by
    its magic number, JSON-FG by its first byte. So a URL with no usable file extension -
    a WFS ``GetFeature`` request, say - is identified from what it served. The extension
    is the fallback, and is all an output target that does not exist yet can offer.

    A shapefile is the ``.shp`` itself, never the directory holding it: GDAL exposes each
    ``.shp`` in a directory as a layer and
    [`ShapeRepository`][xplan_tools.interface.shape.ShapeRepository] reads only the first
    one, so naming the file is what makes the choice deterministic.

    Args:
        source: A file path, or a buffer holding the datasource.
        suffix: File extension to fall back on, e.g. `".gml"`. Ignored when `source` is a
            path, which carries its own.

    Raises:
        ForbiddenDoctypeError: The datasource is an XML document declaring a DTD.
        UnsupportedRootElementError: The datasource is XML, but opens with an element
            this package does not read.

    Returns:
        The repository type, or `None` when the datasource could not be identified.
    """
    return _sniff(source, suffix)[0]

`Datasource(source, *, allow_remote=None, format=None, for_write=False)`

Where a datasource is, what it holds, and which repository reads it.

The value object every repository stores. Construction classifies and validates but never reads: a path is opened, and a remote URL retrieved, only when an attribute first needs the bytes.

Usage example

# read a local file
collection = Datasource("plan.gml").repo().get_all()
# read a WFS response, which carries no usable file extension
ds = Datasource(wfs_url, allow_remote=True)
ds.format         # -> "gml", identified from the retrieved content
ds.repo_class     # -> <class GMLRepository>
collection = ds.repo().get_all()
# write to a GeoPackage; a database is always addressed by connection URL
Datasource("gpkg:///out.gpkg").repo().save_all(collection)
# name the format instead of identifying it from the content
Datasource("out.dat", format="gml").repo().save_all(collection)

Attributes:

Name	Type	Description
`raw`		The datasource exactly as it was handed over.
`kind`	`Kind`	Where the datasource's bytes are.
`allow_remote`		Whether retrieval was permitted; `None` defers to the `XMAS_DS_ALLOW_REMOTE` setting.
`for_write`		Whether this is a write target, and so classified by name.

Classifies and validates a datasource, without reading it.

Parameters:

Name	Type	Description	Default
`source`	`Any`	A file path as a string or `os.PathLike`, a connection URL, an `http(s)` URL, a file-like object, or an existing `Datasource`, whose classification is reused - so handing a resolved datasource to a repository does not resolve it twice.	required
`allow_remote`	`bool \| None`	Whether an `http(s)` datasource may be retrieved; defaults to the `XMAS_DS_ALLOW_REMOTE` setting. Pass `False` for an output target, which is never fetched.	`None`
`format`	`RepoType \| None`	The repository type, when it is known. Skips identifying it.	`None`
`for_write`	`bool`	Whether this datasource is a write target, whose `format` is then taken from its name rather than from its content.	`False`

Raises:

Type	Description
`UnsupportedDatasourceError`	The datasource names a location this package refuses to open, or is remote while remote access is disabled.

Source code in xplan_tools/interface/datasource.py

def __init__(
    self,
    source: Any,
    *,
    allow_remote: bool | None = None,
    format: RepoType | None = None,
    for_write: bool = False,
) -> None:
    """Classifies and validates a datasource, without reading it.

    Args:
        source: A file path as a string or `os.PathLike`, a connection URL, an
            ``http(s)`` URL, a file-like object, or an existing `Datasource`, whose
            classification is reused - so handing a resolved datasource to a
            repository does not resolve it twice.
        allow_remote: Whether an ``http(s)`` datasource may be retrieved; defaults to
            the ``XMAS_DS_ALLOW_REMOTE`` setting. Pass `False` for an output target,
            which is never fetched.
        format: The repository type, when it is known. Skips identifying it.
        for_write: Whether this datasource is a write target, whose
            [`format`][xplan_tools.interface.datasource.Datasource] is then taken
            from its name rather than from its content.

    Raises:
        UnsupportedDatasourceError: The datasource names a location this package
            refuses to open, or is remote while remote access is disabled.
    """
    if isinstance(source, Datasource):
        # already classified; carry the caches over so nothing is fetched twice
        self.__dict__.update(source.__dict__)
        if allow_remote is not None:
            self.allow_remote = allow_remote
        if for_write and not self.for_write:
            # promoted to a write target after the fact: a format identified from
            # the content that is about to be replaced no longer applies
            self.__dict__.pop("format", None)
            self.for_write = True
        if format is not None:
            self.format = format
        # the caches came over too, so a `buffer` fetched under the policy this call
        # just revoked would otherwise be handed back as if it were still allowed
        self._guard_remote(source)
        return
    if isinstance(source, os.PathLike):
        # so that Datasource(other.uri) round-trips for a local file
        source = os.fspath(source)
    self.raw = source
    self.allow_remote = allow_remote
    self.for_write = for_write
    if format is not None:
        # seeds the cached_property below; writing None would shadow it for good
        self.format = format
    self.kind: Kind = _kind_of(source)
    self._guard_remote(source)

`buffer` `cached` `property`

The datasource's bytes, when they are in memory rather than on disk.

Retrieves a remote datasource on first access. A path is left for the repository to open, so a large file is never read into memory here. A stream that cannot be rewound is drained into a BytesIO once, so that identifying the format does not consume it and leave every later reader with nothing.

Raises:

Type	Description
`RemoteFetchError`	A remote datasource could not be retrieved.

`format` `cached` `property`

Which repository reads this datasource, identified from its content.

A write target (for_write) is identified from its name instead: whatever is at the path now is about to be replaced, so it says nothing about what is being written. Reading it would refuse to overwrite a file holding a document this package does not read.

Raises:

Type	Description
`ForbiddenDoctypeError`	The datasource is an XML document declaring a DTD.
`UnsupportedRootElementError`	The datasource is XML, but opens with an element this package does not read.
`RemoteFetchError`	A remote datasource could not be retrieved.

`is_remote` `property`

Whether the datasource is an http(s) URL.

`repo_class` `property`

The repository class that reads this datasource.

Imported on demand, one module at a time: those modules import this one in turn, so the import cannot be made at module scope. One at a time rather than all four because DBRepository drags in Alembic, which a GML file has no use for.

Raises:

Type	Description
`DatasourceError`	The datasource could not be identified.

`root_info` `cached` `property`

The root element of an XML datasource, read without parsing the rest of it.

Seeded by format when the content sniff already read it, so a GML datasource is preflighted once however it was reached.

Raises:

Type	Description
`ForbiddenDoctypeError`	The datasource is an XML document declaring a DTD.
`UnsupportedRootElementError`	The datasource opens with an element this package does not read.
`XMLParseError`	The datasource does not hold well-formed XML.

`source` `property`

What a file-based repository reads: the buffer if there is one, else the path.

DBRepository reads uri instead.

`uri` `cached` `property`

The datasource as a typed location, or None for a file-like object.

A sqlalchemy.URL masks its password, which is why this rather than raw is what gets logged. An http(s) URL is split by the standard library rather than validated by a URL model: classification has already accepted the scheme, and a stricter parse here would raise an error no caller of this package expects.

Raises:

Type	Description
`DatasourceError`	The connection URL could not be parsed. The URL itself is left out of the message, since a password cannot be masked in a string that did not parse.

`open()`

Yields the datasource as a binary stream positioned at the start.

A path is opened and closed here; a buffer is rewound and put back where it was found, so a caller's stream comes back as it was handed over.

Raises:

Type	Description
`DatasourceError`	The datasource is a database connection, which holds no stream of bytes.
`RemoteFetchError`	A remote datasource could not be retrieved.
`OSError`	The path could not be opened.

Yields:

Type	Description
`IO[bytes]`	A binary stream over the datasource.

Source code in xplan_tools/interface/datasource.py

@contextmanager
def open(self) -> Iterator[IO[bytes]]:
    """Yields the datasource as a binary stream positioned at the start.

    A path is opened and closed here; a buffer is rewound and put back where it was
    found, so a caller's stream comes back as it was handed over.

    Raises:
        DatasourceError: The datasource is a database connection, which holds no
            stream of bytes.
        RemoteFetchError: A remote datasource could not be retrieved.
        OSError: The path could not be opened.

    Yields:
        A binary stream over the datasource.
    """
    if self.kind == "db":
        raise DatasourceError(
            f"{self!r} is a database connection, not a byte stream"
        )
    with as_binary_stream(self.source) as stream:
        yield stream

`read_bytes()`

Returns the whole datasource, retrieving a remote one on the way.

Raises:

Type	Description
`DatasourceError`	The datasource is a database connection.
`RemoteFetchError`	A remote datasource could not be retrieved.
`OSError`	The path could not be opened.

Returns:

Type	Description
`bytes`	The datasource's bytes.

Source code in xplan_tools/interface/datasource.py

def read_bytes(self) -> bytes:
    """Returns the whole datasource, retrieving a remote one on the way.

    Raises:
        DatasourceError: The datasource is a database connection.
        RemoteFetchError: A remote datasource could not be retrieved.
        OSError: The path could not be opened.

    Returns:
        The datasource's bytes.
    """
    with self.open() as stream:
        return stream.read()

`repo()`

Builds the repository that reads this datasource.

Raises:

Type	Description
`DatasourceError`	The datasource could not be identified.

Returns:

Type	Description
`BaseRepository`	An instance of the matching repository class.

Source code in xplan_tools/interface/datasource.py

def repo(self) -> BaseRepository:
    """Builds the repository that reads this datasource.

    Raises:
        DatasourceError: The datasource could not be identified.

    Returns:
        An instance of the matching repository class.
    """
    repo_class = self.repo_class
    logger.debug(f"initializing {repo_class.__name__}")
    return repo_class(self)

`write_bytes(data)`

Writes data to the datasource, encoding for a text buffer.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	The serialized document.	required

Raises:

Type	Description
`DatasourceError`	The datasource is not something this package writes to - a database connection, which its repository writes through SQLAlchemy, or a remote URL, which is never written back.
`OSError`	The path could not be written.

Returns:

Type	Description
`int`	The number of bytes, or characters for a text buffer, written.

Source code in xplan_tools/interface/datasource.py

def write_bytes(self, data: bytes) -> int:
    """Writes `data` to the datasource, encoding for a text buffer.

    Args:
        data: The serialized document.

    Raises:
        DatasourceError: The datasource is not something this package writes to - a
            database connection, which its repository writes through SQLAlchemy, or
            a remote URL, which is never written back.
        OSError: The path could not be written.

    Returns:
        The number of bytes, or characters for a text buffer, written.
    """
    match self.kind:
        case "path":
            written = Path(self.raw).write_bytes(data)
        case "buffer":
            written = self._replace_buffer(
                data.decode("utf-8")
                if isinstance(self.raw, io.TextIOBase)
                else data
            )
        case _:
            raise DatasourceError(f"{self!r} cannot be written to")
    # Whatever was read out of the datasource describes what used to be there. A text
    # buffer is the one kind where that is not self-correcting: `buffer` holds an
    # encoded snapshot rather than `raw` itself, so without this a read after a write
    # returns the document from before it. `format` is deliberately kept - it is the
    # routing decision the caller made, not an observation of the content.
    self.__dict__.pop("buffer", None)
    self.__dict__.pop("root_info", None)
    return written

`BaseRepository(datasource)`

Bases: Protocol

Protocol defining the common interface for all repository implementations.

Initialize the repository from a datasource.

Source code in xplan_tools/interface/base.py

def __init__(self, datasource: Any) -> None:
    """Initialize the repository from a datasource."""
    ...

`delete(obj_id, *, session=None)`

Delete a BaseFeature.

Source code in xplan_tools/interface/base.py

def delete(self, obj_id: UUID, *, session: object = None) -> BaseFeature:
    """Delete a BaseFeature."""
    ...

`delete_plan_by_id(plan_id, *, session=None)`

Delete a plan object with its related features.

Source code in xplan_tools/interface/base.py

def delete_plan_by_id(self, plan_id: UUID, *, session: object = None) -> None:
    """Delete a plan object with its related features."""
    ...

`get(obj_id, *, session=None)`

Get a specific BaseFeature by id.

Source code in xplan_tools/interface/base.py

def get(self, obj_id: UUID, *, session: object = None) -> BaseFeature:
    """Get a specific BaseFeature by id."""
    ...

`get_all()`

Get all BaseFeatures.

Source code in xplan_tools/interface/base.py

def get_all(self) -> BaseCollection:
    """Get all BaseFeatures."""
    ...

`get_plan_by_id(plan_id, *, session=None)`

Get a plan object with its related features.

Source code in xplan_tools/interface/base.py

def get_plan_by_id(
    self, plan_id: UUID, *, session: object = None
) -> BaseCollection:
    """Get a plan object with its related features."""
    ...

`patch(obj_id, partial_obj, *, session=None)`

Partially update a BaseFeature.

Source code in xplan_tools/interface/base.py

def patch(
    self, obj_id: UUID, partial_obj: dict, *, session: object = None
) -> BaseFeature:
    """Partially update a BaseFeature."""
    ...

`save(obj, *, session=None)`

Store a BaseFeature.

Source code in xplan_tools/interface/base.py

def save(self, obj: BaseFeature, *, session: object = None) -> None:
    """Store a BaseFeature."""
    ...

`save_all(features, *, session=None)`

Store a BaseCollection.

Source code in xplan_tools/interface/base.py

def save_all(self, features: BaseCollection, *, session: object = None) -> None:
    """Store a BaseCollection."""
    ...

`update(obj_id, new_obj, *, session=None)`

Update a BaseFeature.

Source code in xplan_tools/interface/base.py

def update(
    self, obj_id: UUID, new_obj: BaseFeature, *, session: object = None
) -> BaseFeature:
    """Update a BaseFeature."""
    ...

`GMLRepository(datasource)`

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 \| BytesIO \| StringIO \| Datasource`	A file path as a String, a file-like object or a `Datasource`. An `http(s)` URL is retrieved on first read, but only when remote access is enabled.	required

Raises:

Type	Description
`UnsupportedDatasourceError`	The datasource is one this package refuses to open, or is remote while remote access is disabled.

Source code in xplan_tools/interface/gml.py

def __init__(
    self,
    datasource: str | io.BytesIO | io.StringIO | Datasource,
) -> None:
    """Initializes the GML Repository.

    Args:
        datasource: A file path as a String, a file-like object or a
            [`Datasource`][xplan_tools.interface.datasource.Datasource]. An
            ``http(s)`` URL is retrieved on first read, but only when remote access
            is enabled.

    Raises:
        UnsupportedDatasourceError: The datasource is one this package refuses to
            open, or is remote while remote access is disabled.
    """
    self.datasource = Datasource(datasource)
    self.appschema: Appschema | None = None

`content` `cached` `property`

The parsed XML tree.

`root_info` `property`

The root element's name, namespace bindings and xsi:schemaLocation.

Read by streaming to the first element only, so a document this package cannot read is rejected before the full parse. Identifying the datasource's format already read it, so this is normally the cached result of that one read.

`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`
`**kwargs`	`dict`	Read options. `context` is handed to the collection's validation as its pydantic validation context: setting `DROP_INVALID_REFS` in it makes unresolvable and external references be dropped rather than raise, and the dropped `InvalidReference`s are collected in the same dict under `INVALID_REFS` for the caller to read back. It does not cover a malformed `xlink:href` - one that is neither intra-document (`#GML_<uuid>`, `urn:uuid:<uuid>`) nor an absolute URI - which fails while the feature itself is read and so never reaches the collection.	`{}`

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.
        **kwargs: Read options. `context` is handed to the collection's validation
            as its pydantic validation context: setting `DROP_INVALID_REFS` in it
            makes unresolvable and external references be dropped rather than raise,
            and the dropped `InvalidReference`s are collected in the same dict under
            `INVALID_REFS` for the caller to read back. It does not cover a malformed
            `xlink:href` - one that is neither intra-document (`#GML_<uuid>`,
            `urn:uuid:<uuid>`) nor an absolute URI - which fails while the feature
            itself is read and so never reaches the collection.
    """
    self.appschema = self._get_appschema()

    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")
            # a local fragment names the whole gml:id, a "urn:uuid:" only the UUID
            # within it - `id_mapping` holds a key for either shape
            if href.startswith("#"):
                key = href[1:]
            elif href.startswith("urn:uuid:") and (uuid := parse_uuid(href)):
                key = str(uuid)
            else:
                continue
            if new_id := id_mapping.get(key, 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")
        # A gml:id is an xsd:ID and so unique within the document. Where it is not,
        # both "#<gml_id>" and "urn:uuid:" references to it name two features at
        # once, and nothing in the document says which was meant - unlike two
        # distinct ids that merely carry the same UUID, handled below.
        if gml_id in seen_ids:
            raise DuplicateGmlIdError(
                f"gml:id '{gml_id}' is used by more than one feature; a gml:id is "
                "unique within a document, and a reference to a repeated one names "
                "two features at once"
            )
        if gml_id:
            seen_ids.add(gml_id)
        uuid = parse_uuid(gml_id)
        if (
            not uuid
            or 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
            # Only where this UUID is being retired document-wide. A feature renamed
            # because another one - under a different gml:id - already claimed the
            # UUID must not capture that one's "urn:uuid:" references: the feature
            # that kept its id is the one still answering to them.
            if uuid and always_generate_ids:
                id_mapping[str(uuid)] = new_id
            logger.info(f"GML ID '{gml_id}' replaced with UUIDv4 '{new_id}'")
        else:
            collection[uuid] = "placeholder"

    root: etree._Element = 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 SRSNotFoundError("No SRS could be found")
    else:
        srid = parse_srs(srs)

    collection = {}
    id_mapping = {}
    seen_ids: set[str] = set()

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

`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.	`{}`

Raises:

Type	Description
`AppschemaNotFoundError`	The collection's application schema has no GML encoding in this package.

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.

    Raises:
        AppschemaNotFoundError: The collection's application schema has no GML
            encoding in this package.
    """
    self.appschema = features.appschema

    match self.appschema.prefix:
        case "xplan":
            nsmap = {
                None: str(self.appschema.namespace_uri).rstrip("/"),
                "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.appschema.version}/XPlanung-Operationen.xsd",
                    "{http://www.opengis.net/gml/3.2}id": f"GML_{uuid4()}",
                },
                nsmap=nsmap,
            )
        case "xtrasse":
            nsmap = {
                None: str(self.appschema.namespace_uri).rstrip("/"),
                "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/{self.appschema.version}/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,
            )
        case "xwp":
            nsmap = {
                None: str(self.appschema.namespace_uri).rstrip("/"),
                "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,
            )
        case "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,
            )

        case _:
            raise AppschemaNotFoundError(
                f"cannot write GML for appschema prefix {self.appschema.prefix!r}"
            )

    if self.appschema.prefix not in ["xtrasse", "xwp"]:
        bounds = etree.SubElement(
            root,
            (
                "{http://www.opengis.net/gml/3.2}boundedBy"
                if self.appschema.prefix == "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)
            etree.SubElement(
                root,
                (
                    "{http://www.opengis.net/gml/3.2}featureMember"
                    if self.appschema.prefix == "xplan"
                    else (
                        "{http://www.opengis.net/ogcapi-features-1/1.0/sf}featureMember"
                        if self.appschema.prefix 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": format_srs(
            features.srid, "url" if self.appschema.prefix == "plu" else "short"
        )
    }

    if self.appschema.prefix 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.prefix == "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)`

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 \| Datasource`	A file path as a String, a file-like object or a `Datasource`. An `http(s)` URL is retrieved on first read, but only when remote access is enabled.	required

Raises:

Type	Description
`UnsupportedDatasourceError`	The datasource is one this package refuses to open, or is remote while remote access is disabled.

Source code in xplan_tools/interface/jsonfg.py

def __init__(
    self,
    datasource: str | IO | Datasource,
) -> None:
    """Initializes the JSON-FG Repository.

    Args:
        datasource: A file path as a String, a file-like object or a
            [`Datasource`][xplan_tools.interface.datasource.Datasource]. An
            ``http(s)`` URL is retrieved on first read, but only when remote access
            is enabled.

    Raises:
        UnsupportedDatasourceError: The datasource is one this package refuses to
            open, or is remote while remote access is disabled.
    """
    self.datasource = Datasource(datasource)
    self.appschema: Appschema | None = None

`content` `property`

The JSON data as a dict.

Raises:

Type	Description
`JsonFGParseError`	The datasource does not hold readable JSON. A truncated response is reported here rather than surfacing later as a missing schema link.
`DatasourceError`	The datasource holds no readable stream of bytes.

`get_all(**kwargs)`

Retrieves a Feature Collection to the datasource.

Parameters:

Name	Type	Description	Default
`**kwargs`	`dict`	Read options. `context` is handed to the collection's validation as its pydantic validation context: setting `DROP_INVALID_REFS` in it makes unresolvable and external references be dropped rather than raise, and the dropped `InvalidReference`s are collected in the same dict under `INVALID_REFS` for the caller to read back.	`{}`

Source code in xplan_tools/interface/jsonfg.py

def get_all(self, **kwargs: dict) -> BaseCollection:
    """Retrieves a Feature Collection to the datasource.

    Args:
        **kwargs: Read options. `context` is handed to the collection's validation
            as its pydantic validation context: setting `DROP_INVALID_REFS` in it
            makes unresolvable and external references be dropped rather than raise,
            and the dropped `InvalidReference`s are collected in the same dict under
            `INVALID_REFS` for the caller to read back.
    """
    self.appschema = self._get_appschema()

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

`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

    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:
        # one document per feature type, so the target names a family of files rather
        # than a single one - which only a path can do
        if self.datasource.kind != "path":
            raise DatasourceError(
                f"{self.datasource!r} cannot be fanned out by feature type: "
                "one document per feature type needs a file path to write to"
            )
        featuretypes = {}
        srid = features.srid
        for feature in features.features.values():
            featuretypes.setdefault(feature.get_name(), []).append(feature)
        for featuretype, typed_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 typed_features
            )
            self._write_to_datasource(
                collection, self._fan_out_datasource(featuretype)
            )

`DBRepository(datasource)`

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 \| URL \| Datasource`	A connection uri, a `sqlalchemy.URL` or a `Datasource`.	required

Source code in xplan_tools/interface/db.py

def __init__(
    self,
    datasource: str | URL | Datasource,
) -> 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 uri, a `sqlalchemy.URL` or a
            [`Datasource`][xplan_tools.interface.datasource.Datasource].
    """
    settings = get_settings()
    self.datasource = Datasource(datasource)
    if self.datasource.kind != "db":
        # a bare path sniffs as "db" from its .gpkg/.sqlite suffix or its SQLite
        # magic number, but SQLAlchemy needs a scheme to pick a dialect
        raise DatasourceError(
            f"{self.datasource.raw!r} is not a database connection URL; address a "
            "GeoPackage as gpkg:///<file> and a SQLite database as sqlite:///<file>"
        )
    # narrowed once here: `uri` is typed for every kind of datasource, and only the
    # check above makes it a connection URL
    self.url: URL = self.datasource.uri
    self.content = None
    self.schema = settings.db_schema
    self.srid = settings.db_srid
    self.dialect = self.url.get_dialect().name
    self.Session = sessionmaker(bind=self._engine, expire_on_commit=False)

    self.alembic_cfg = config.Config()
    self.alembic_cfg.set_main_option(
        "script_location", "xplan_tools:model:migrations"
    )
    self._ensure_repo()

`create_tables()`

Creates coretable and related/spatial tables in the database.

Source code in xplan_tools/interface/db.py

def create_tables(self) -> None:
    """Creates coretable and related/spatial tables in the database."""

    @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}")
    # create_tables is only ever called for file-based (non-Postgres) DBs, so
    # drop navigable_roles_config (seeded/used only by the Postgres migration)
    # and the geopackage-only gpkgext_relations for non-geopackage dialects.
    tables = [
        table
        for table in Base.metadata.sorted_tables
        if table.name != "navigable_roles_config"
        and (self.dialect == "geopackage" or table.name != "gpkgext_relations")
    ]
    tables[0].append_column(  # tables[0] is still coretable
        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":
        with self._engine.connect() as conn:
            self.alembic_cfg.attributes["connection"] = conn
            command.downgrade(self.alembic_cfg, "base")
    else:
        Base.metadata.drop_all(self._engine)

`get_session(session=None)`

Yield a managed session.

Source code in xplan_tools/interface/db.py

@contextmanager
def get_session(
    self, session: Session | None = None
) -> Generator[Session, None, None]:
    """Yield a managed session."""
    if session is not None:
        yield session
    else:
        with self.Session() as _session:
            with _session.begin():
                yield _session

`AsyncDBRepository(datasource)`

Bases: DBRepository

Async PostgreSQL-backed repository that reuses synchronous migrations.

Source code in xplan_tools/interface/db.py

def __init__(self, datasource: str | URL | Datasource) -> None:
    super().__init__(datasource=datasource)
    self._engine.dispose()
    if self.dialect != "postgresql":
        raise ValueError(
            "Async DBRepository is only supported for PostgreSQL datasources"
        )

    self.Session: async_sessionmaker[AsyncSession] = async_sessionmaker(
        bind=self._async_engine,
        expire_on_commit=False,
    )

`get_session(session=None)` `async`

Yield a managed session.

Source code in xplan_tools/interface/db.py

@asynccontextmanager
async def get_session(
    self, session: AsyncSession | None = None
) -> AsyncGenerator[AsyncSession, None]:
    """Yield a managed session."""
    if session is not None:
        yield session
    else:
        async with self.Session() as _session:
            async with _session.begin():
                yield _session

`ShapeRepository(datasource)`

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.

Initializes the Shape Repository.

Parameters:

Name	Type	Description	Default
`datasource`	`str \| Datasource`	A shapefile - the `.shp` itself - as a String or a `Datasource`. Its sidecars (`.dbf`, `.shx`, `.prj`) are picked up by GDAL.	required

Raises:

Type	Description
`UnsupportedDatasourceError`	The datasource is one this package refuses to open. GDAL reads the datasource here, so a path that names nowhere on this filesystem - a `/vsicurl/` or `/vsizip/` location, say - is refused rather than handed on.
`DatasourceError`	The datasource is not a path. GDAL opens a shapefile by name, alongside its sidecars, so a buffer or a URL cannot be one.

Source code in xplan_tools/interface/shape.py

def __init__(
    self,
    datasource: str | Datasource,
) -> None:
    """Initializes the Shape Repository.

    Args:
        datasource: A shapefile - the ``.shp`` itself - as a String or a
            [`Datasource`][xplan_tools.interface.datasource.Datasource]. Its
            sidecars (``.dbf``, ``.shx``, ``.prj``) are picked up by GDAL.

    Raises:
        UnsupportedDatasourceError: The datasource is one this package refuses to
            open. GDAL reads the datasource here, so a path that names nowhere on
            this filesystem - a ``/vsicurl/`` or ``/vsizip/`` location, say - is
            refused rather than handed on.
        DatasourceError: The datasource is not a path. GDAL opens a shapefile by
            name, alongside its sidecars, so a buffer or a URL cannot be one.
    """
    self.datasource = Datasource(datasource)
    if self.datasource.kind != "path":
        # a shapefile buffer identifies itself by its magic number, but GDAL takes a
        # name, and the sidecars a shapefile needs are not in the buffer either
        raise DatasourceError(
            f"{self.datasource!r} is not a shapefile path; a shapefile is read from "
            "disk together with its .dbf/.shx sidecars, so it cannot be read from a "
            "buffer or a URL"
        )
    self.appschema = Appschema.from_prefix("xplan", "6.0")

`get_all(**kwargs)`

Retrieves a Feature Collection from the datasource.

Parameters:

Name	Type	Description	Default
`**kwargs`		Read options. `context` is handed to the collection's validation as its pydantic validation context: setting `DROP_INVALID_REFS` in it makes unresolvable and external references be dropped rather than raise, and the dropped `InvalidReference`s are collected in the same dict under `INVALID_REFS` for the caller to read back.	`{}`

Source code in xplan_tools/interface/shape.py

def get_all(self, **kwargs) -> BaseCollection:
    """Retrieves a Feature Collection from the datasource.

    Args:
        **kwargs: Read options. `context` is handed to the collection's validation
            as its pydantic validation context: setting `DROP_INVALID_REFS` in it
            makes unresolvable and external references be dropped rather than raise,
            and the dropped `InvalidReference`s are collected in the same dict under
            `INVALID_REFS` for the caller to read back.
    """
    schema = self.appschema

    def _get_date(date_field: str):
        try:
            value = ft.GetField(date_field)
            return str(parse(value).date())
        except Exception:
            return None

    def _get_field(entry: str):
        try:
            return ft.GetField(entry)
        except Exception:
            return None

    def _add_ref(external_ref: ExternalReferenceUtil, typ: str) -> None:
        ref = {
            "referenzName": "Unbekannt",
            "referenzURL": str(external_ref.ref_url),
            "typ": typ,
        }
        if external_ref.georef_url:
            if not kwargs.get("ref_check", None) or external_ref.georef_url_valid():
                ref["georefURL"] = str(external_ref.georef_url)
            else:
                logger.warning(
                    "Skipping invalid georefURL: %s", external_ref.georef_url
                )

        if not kwargs.get("ref_check", None) or external_ref.ref_url_valid():
            data.setdefault("externeReferenz", []).append(ref)
        else:
            raise FileNotFoundError(
                f"File {external_ref.ref_url} could not be added."
            )

    planart_mapping = {
        "BP": {
            "1000": "10000",
            "2000": "10001",
            "3000": "3000",
            "4000": "40001",
            "5000": "40002",
            "6000": "1000",
            "7000": "5000",
            "8000": "40000",
        },
        "FP": {
            "1000": "1000",
            "2000": "2000",
            "3000": "3000",
            "4000": "4000",
        },
    }

    typ_mapping = {
        "BESCHRURL": "1000",
        "BEGRURL": "1010",
        "LEGENDEURL": "1020",
        "RECHTSURL": "1030",
        "LIEGURL": "1040",
        "UMWELTBERI": "1050",
        "TEXTURL": "9998",
    }

    if not kwargs.get("raster_as_refscan", None):
        typ_mapping["SCANURL"] = "1070"

    collection = {}

    with gdal.OpenEx(
        self.datasource.source, allowed_drivers=["ESRI Shapefile"]
    ) as ds:
        lyr = ds.GetLayer(0)
        spatial_ref = lyr.GetSpatialRef()
        srid = None

        for ft in lyr:
            try:
                if srid is None:
                    # Prefer the layer's SRS, falling back to the feature's;
                    # derive the EPSG code, identifying it from the WKT if it
                    # carries no authority. AutoIdentifyEPSG raises on failure.
                    sr = spatial_ref or ft.geometry().GetSpatialReference()
                    code = sr.GetAuthorityCode(None)
                    if not code:
                        try:
                            sr.AutoIdentifyEPSG()
                            code = sr.GetAuthorityCode(None)
                        except RuntimeError:
                            code = None
                    srid = int(code) if code and code.isdigit() else None

                geom = s.from_wkb(bytes(ft.geometry().ExportToWkb())).reverse()
                data = {
                    "id": uuid4(),
                    "name": _get_field(kwargs.get("name_field", "PLANID")),
                    "nummer": _get_field("NUMMER"),
                    "beschreibung": _get_field("BESCHR"),
                    "kommentar": _get_field("KOMMENTAR"),
                    "aufstellungsbeschlussDatum": _get_date("AUFSTELLUN"),
                    "aenderungenBisDatum": _get_date("AENDERUNGE"),
                    "technHerstellDatum": _get_date("DATHERST"),
                    "untergangsDatum": _get_date("DATUNTER"),
                    "auslegungsStartDatum": [date]
                    if (date := _get_date("AUSLEGUNGS"))
                    else None,
                    "traegerbeteiligungsStartDatum": [date]
                    if (date := _get_date("TRAEGERBET"))
                    else None,
                    "rechtsstand": str(_get_field("RECHTSSTA")),
                    "erstellungsMassstab": _get_field("ERSTELLUNG"),
                    "gemeinde": [
                        {
                            "ags": f"0{_get_field('GKZ')}",
                            "gemeindeName": _get_field("STADT"),
                        }
                    ],
                    "raeumlicherGeltungsbereich": {
                        "srid": srid,
                        "wkt": s.to_wkt(geom),
                    },
                    "hoehenbezug": _get_field("HOEHENBEZU"),
                }
            except AttributeError as e:
                logger.error(
                    f"GEOM could not be parsed for {ft}: {e}", exc_info=True
                )
                raise e
            except KeyError as e:
                logger.error(f"Missing Field for {ft}: {e}", exc_info=True)
                raise e

            for field in ["NAME", "PLANID", "ROKNR"]:
                if value := _get_field(field):
                    data.setdefault("hatGenerAttribut", []).append(
                        {
                            "name": f"shp:{field}",
                            "wert": str(value),
                        }
                    )

            # Set the plan_type to BP or FP, depending on which of the two date attributes
            # INKRAFTTRE or WIRKSAMKEI is found in ft
            if inkraft := _get_date("INKRAFTTRE"):
                data["inkrafttretensDatum"] = inkraft
                data["rechtsverordnungsDatum"] = _get_date("RECHTSVERO")
                data["satzungsbeschlussDatum"] = _get_date("SATZUNGSBE")
                plan_type = "BP"
            elif wirksam := _get_date("WIRKSAMKEI"):
                data["wirksamkeitsDatum"] = wirksam
                data["planbeschlussDatum"] = _get_date("PLANBESCHL")
                data["entwurfsbeschlussDatum"] = _get_date("ENTWURFBES")
                plan_type = "FP"
            else:
                error_message = "Required dates 'INKRAFTTRE' (BP) or 'WIRKSAMKEI' (FP) are missing"
                logger.error(error_message)
                raise KeyError(error_message)

            plan_art_entry = planart_mapping.get(plan_type).get(
                str(_get_field("PLANART")), None
            )
            if not plan_art_entry:
                error_message = "No mapping found for required attribute 'PLANART'"
                logger.error(error_message)
                raise KeyError(error_message)
            data["planArt"] = (
                [plan_art_entry] if plan_type == "BP" else plan_art_entry
            )

            if aendert := _get_field("AENDERID"):
                data["aendertPlan"] = [
                    {"planName": aendert, "aenderungsArt": "1000"}
                ]

            for field, typ in typ_mapping.items():
                if url := _get_field(field):
                    try:
                        _add_ref(
                            ExternalReferenceUtil(
                                url,
                                _get_field("LIEGGEOREF") if typ == "1040" else None,
                            ),
                            typ,
                        )
                    except ValidationError:
                        logger.warning(
                            f"URL {url} for field {field} not valid: skipping"
                        )
                    except FileNotFoundError:
                        logger.warning(
                            f"URL {url} for field {field} of plan {data['name']} could not be added: skipping"
                        )

            if kwargs.get("raster_as_refscan", None):
                if ref_url := _get_field("SCANURL"):
                    try:
                        raster_ref = RasterReferenceUtil(
                            ref_url, _get_field("GEOREFURL")
                        )
                    except ValidationError:
                        logger.info(f"Raster URL {url} not valid: skip")
                    else:
                        if (
                            not kwargs.get("ref_check", None)
                            or raster_ref.raster_data_valid()
                        ):
                            bereich_data = {
                                "id": uuid4(),
                                "nummer": 0,
                                "gehoertZuPlan": data.get("id"),
                                "refScan": [
                                    {
                                        "referenzName": "Unbekannt",
                                        "referenzURL": str(raster_ref.ref_url),
                                        "art": "PlanMitGeoreferenz",
                                        "georefURL": str(raster_ref.georef_url)
                                        if raster_ref.georef_url
                                        else None,
                                    }
                                ],
                            }
                            bereich = schema.model_factory(
                                f"{plan_type}_Bereich"
                            ).model_validate(bereich_data)
                            data.setdefault("bereich", [])
                            data["bereich"].append(bereich_data["id"])
                            collection[bereich.id] = bereich
                        else:
                            logger.warning(
                                f"URL {url} for field SCANURL of plan {data['name']} could not be added: skipping"
                            )

            plan = schema.model_factory(f"{plan_type}_Plan").model_validate(data)
            collection[plan.id] = plan

        if not srid:
            raise ValueError("Could not identify SRID")

    return BaseCollection.from_features(
        collection, int(srid), self.appschema, context=kwargs.get("context")
    )

Interface

interface

Datasource(source, *, allow_remote=None, format=None, for_write=False)

buffer cached property

format cached property

is_remote property

repo_class property

root_info cached property

source property

uri cached property

open()

read_bytes()

repo()

write_bytes(data)

fetch_remote(url, *, allow_remote=None, timeout=None, max_bytes=None)

repo_factory(datasource='', repo_type=None, allow_remote=None)

sniff_datasource(source, *, suffix='')

Datasource(source, *, allow_remote=None, format=None, for_write=False)

buffer cached property

format cached property

is_remote property

repo_class property

root_info cached property

source property

uri cached property

open()

read_bytes()

repo()

write_bytes(data)

BaseRepository(datasource)

delete(obj_id, *, session=None)

delete_plan_by_id(plan_id, *, session=None)

get(obj_id, *, session=None)

get_all()

get_plan_by_id(plan_id, *, session=None)

patch(obj_id, partial_obj, *, session=None)

save(obj, *, session=None)

save_all(features, *, session=None)

update(obj_id, new_obj, *, session=None)

GMLRepository(datasource)

content cached property

root_info property

get_all(always_generate_ids=False, **kwargs)

save_all(features, **kwargs)

JsonFGRepository(datasource)

content property

get_all(**kwargs)

save_all(features, **kwargs)

DBRepository(datasource)

create_tables()

delete_tables()

get_session(session=None)

AsyncDBRepository(datasource)

get_session(session=None) async

ShapeRepository(datasource)

get_all(**kwargs)

`interface`

`Datasource(source, *, allow_remote=None, format=None, for_write=False)`

`buffer` `cached` `property`

`format` `cached` `property`

`is_remote` `property`

`repo_class` `property`

`root_info` `cached` `property`

`source` `property`

`uri` `cached` `property`

`open()`

`read_bytes()`

`repo()`

`write_bytes(data)`

`fetch_remote(url, *, allow_remote=None, timeout=None, max_bytes=None)`

`repo_factory(datasource='', repo_type=None, allow_remote=None)`

`sniff_datasource(source, *, suffix='')`

`Datasource(source, *, allow_remote=None, format=None, for_write=False)`

`buffer` `cached` `property`

`format` `cached` `property`

`is_remote` `property`

`repo_class` `property`

`root_info` `cached` `property`

`source` `property`

`uri` `cached` `property`

`open()`

`read_bytes()`

`repo()`

`write_bytes(data)`

`BaseRepository(datasource)`

`delete(obj_id, *, session=None)`

`delete_plan_by_id(plan_id, *, session=None)`

`get(obj_id, *, session=None)`

`get_all()`

`get_plan_by_id(plan_id, *, session=None)`

`patch(obj_id, partial_obj, *, session=None)`

`save(obj, *, session=None)`

`save_all(features, *, session=None)`

`update(obj_id, new_obj, *, session=None)`

`GMLRepository(datasource)`

`content` `cached` `property`

`root_info` `property`

`get_all(always_generate_ids=False, **kwargs)`

`save_all(features, **kwargs)`

`JsonFGRepository(datasource)`

`content` `property`

`get_all(**kwargs)`

`save_all(features, **kwargs)`

`DBRepository(datasource)`

`create_tables()`

`delete_tables()`

`get_session(session=None)`

`AsyncDBRepository(datasource)`

`get_session(session=None)` `async`

`ShapeRepository(datasource)`

`get_all(**kwargs)`