winterthur.roadwork

Exceptions

RoadworkError

Common base class for all non-exit exceptions.

RoadworkConnectionError

Common base class for all non-exit exceptions.

Classes

RoadworkConfig

Looks at ~/.pdb.secret and /etc/pdb.secret (in this order), to extract

RoadworkClient

A proxy to Winterthur's internal roadworks service. Uses redis as

RoadworkCollection

Roadwork

Module Contents

exception winterthur.roadwork.RoadworkError[source]

Bases: Exception

Common base class for all non-exit exceptions.

exception winterthur.roadwork.RoadworkConnectionError[source]

Bases: RoadworkError

Common base class for all non-exit exceptions.

class winterthur.roadwork.RoadworkConfig(hostname: str | None, endpoint: str | None, username: str | None, password: str | None)[source]

Looks at ~/.pdb.secret and /etc/pdb.secret (in this order), to extract the configuration used for the RoadworkClient class.

The configuration is as follows:

HOSTNAME: pdb.example.org
ENDPOINT: 127.0.0.1:6004
USERNAME: username
PASSWORD: password
  • The HOSTNAME is the address of the PDB service.

  • The ENDPOINT is the optional address of the tcp-proxy used.

  • The USERNAME is the NTLM password.

  • The PASSWORD is the NTLM password.

classmethod lookup_paths() Iterator[Path][source]
classmethod lookup() typing_extensions.Self[source]
classmethod parse(path: pathlib.Path) dict[str, str | None][source]
class winterthur.roadwork.RoadworkClient(cache: onegov.core.cache.RedisCacheRegion, hostname: str, username: str, password: str, endpoint: str | None = None)[source]

A proxy to Winterthur’s internal roadworks service. Uses redis as a caching mechanism to ensure performance and reliability.

Since the roadworks service can only be reached inside Winterthur’s network, we rely on a proxy connection during development/testing.

To not expose any information unwittingly to the public, the description of how to connect to that proxy is kept at docs.seantis.ch.

curl() pycurl.Curl[source]
url(path: str) str[source]
get(path: str, lifetime: float = 5 * 60, downtime: float = 60 * 60) Any[source]

Requests the given path, returning the resulting json if successful.

A cache is used in two stages:

  • At the lifetime stage, the cache is returned unconditionally.

  • At the end of the lifetime, the cache is refreshed if possible.

  • At the end of the downtime stage the cache forcefully refreshed.

During its lifetime the object is basically up to 5 minutes out of date. But since the backend may not be available when that time expires we operate with a downtime that is higher (1 hour).

This means that a downtime in the backend will not result in evicted caches, even if the lifetime is up. Once the downtime limit is up we do however evict the cache forcefully, raising an error if we cannot connect to the backend.

get_uncached(path: str) tuple[int, Any][source]
is_cacheable(response: tuple[int, Any]) bool[source]
class winterthur.roadwork.RoadworkCollection(client: RoadworkClient, letter: str | None = None, query: str | None = None)[source]
property letters: list[str][source]
by_filter(filter: str) list[Roadwork][source]
property roadwork: list[Roadwork][source]
by_id(id: int) Roadwork | None[source]
by_letter(letter: str | None) typing_extensions.Self[source]
class winterthur.roadwork.Roadwork(data: dict[str, Any])[source]
property id: int[source]
property letters: Iterator[str][source]
property title: str[source]
property sections: list[typing_extensions.Self][source]
__getitem__(key: str) Any[source]
__contains__(key: str) bool[source]