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

Bases: Exception

Common base class for all non-exit exceptions.

exception winterthur.roadwork.RoadworkConnectionError

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)

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.

hostname

endpoint

username

password

classmethod lookup_paths() → Iterator[Path]

classmethod lookup() → Self

classmethod parse(path: pathlib.Path) → dict[str, str | None]

class winterthur.roadwork.RoadworkClient(cache: onegov.core.cache.RedisCacheRegion, hostname: str, username: str, password: str, endpoint: str | None = None)

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.

cache

hostname

username

password

endpoint

property curl: pycurl.Curl

url(path: str) → str

get(path: str, lifetime: float = 5 * 60, downtime: float = 60 * 60) → Any

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]

is_cacheable(response: tuple[int, Any]) → bool

class winterthur.roadwork.RoadworkCollection(client: RoadworkClient, letter: str | None = None, query: str | None = None)

client

query = None

letter = None

property letters: list[str]

by_filter(filter: str) → list[Roadwork]

property roadwork: list[Roadwork]

by_id(id: int) → Roadwork | None

by_letter(letter: str | None) → Self

class winterthur.roadwork.Roadwork(data: dict[str, Any])

data

convertors

property id: int

property letters: Iterator[str]

property title: str

property sections: list[Self]

__getitem__(key: str) → Any

__contains__(key: str) → bool