init

Module to store the common classes and functions for all cities.

CityData (BaseModel, ABC) pydantic-model

City data abstract class.

Abstracts the city data retrieval.

Source code in now8_api/service/city_data/__init__.py

class CityData(BaseModel, ABC):
    """City data abstract class.

    Abstracts the city data retrieval.
    """

    @abstractmethod
    async def get_estimations(
        self,
        stop: Stop,
    ) -> List[VehicleEstimation]:
        """Return ETA for the next vehicles to the stop.

        Arguments:
            stop: Stop to get the next vehicle arrival estimations for.

        Returns:
            ETA for the next vehicles to the stop.

        Raises:
            NotImplementedError: If the method is not implemented for
                the stop transport type in this city.
            UpstreamError: If the upstream city API fails.
        """

    @abstractmethod
    async def get_stops_city(
        self,
        transport_types: List[TransportType] = None,
    ) -> List[Stop]:
        """Return all the stops of the selected transport types.

        Arguments:
            transport_types: Transport types to get the stops for.
                If none are passed, all of the available ones for the
                city will be included.

        Returns:
            All the stops of the selected transport types.

        Raises:
            NotImplementedError: If the method is not implemented for
                the stop transport type in this city.
        """

    @abstractmethod
    async def get_stops_route(
        self,
        route: Route,
    ) -> Tuple[List[Stop], List[Stop]]:
        """Return all the stops of the selected route.

        Arguments:
            route: Transport route to get the stops for.

        Returns:
            Stops of the selected route for both ways in order.

        Raises:
            NotImplementedError: If the method is not implemented for
                the route transport type in this city.
        """

    @abstractmethod
    async def get_routes_stop(
        self,
        stop: Stop,
    ) -> List[Route]:
        """Return all the routes that pass through the stop.

        Arguments:
            stop: Transport stop to get the routes for.

        Returns:
            Routes that pass through the stop.

        Raises:
            NotImplementedError: If the method is not implemented for
                the stop transport type in this city.
        """

get_estimations(self, stop) async

Return ETA for the next vehicles to the stop.

Parameters:

Name	Type	Description	Default
stop	Stop	Stop to get the next vehicle arrival estimations for.	required

Returns:

Type	Description
List[now8_api.domain.VehicleEstimation]	ETA for the next vehicles to the stop.

Exceptions:

Type	Description
NotImplementedError	If the method is not implemented for the stop transport type in this city.
UpstreamError	If the upstream city API fails.

Source code in now8_api/service/city_data/__init__.py

@abstractmethod
async def get_estimations(
    self,
    stop: Stop,
) -> List[VehicleEstimation]:
    """Return ETA for the next vehicles to the stop.

    Arguments:
        stop: Stop to get the next vehicle arrival estimations for.

    Returns:
        ETA for the next vehicles to the stop.

    Raises:
        NotImplementedError: If the method is not implemented for
            the stop transport type in this city.
        UpstreamError: If the upstream city API fails.
    """

get_routes_stop(self, stop) async

Return all the routes that pass through the stop.

Parameters:

Name	Type	Description	Default
stop	Stop	Transport stop to get the routes for.	required

Returns:

Type	Description
List[now8_api.domain.Route]	Routes that pass through the stop.

Exceptions:

Type	Description
NotImplementedError	If the method is not implemented for the stop transport type in this city.

Source code in now8_api/service/city_data/__init__.py

@abstractmethod
async def get_routes_stop(
    self,
    stop: Stop,
) -> List[Route]:
    """Return all the routes that pass through the stop.

    Arguments:
        stop: Transport stop to get the routes for.

    Returns:
        Routes that pass through the stop.

    Raises:
        NotImplementedError: If the method is not implemented for
            the stop transport type in this city.
    """

get_stops_city(self, transport_types=None) async

Return all the stops of the selected transport types.

Parameters:

Name	Type	Description	Default
transport_types	List[now8_api.domain.TransportType]	Transport types to get the stops for. If none are passed, all of the available ones for the city will be included.	None

Returns:

Type	Description
List[now8_api.domain.Stop]	All the stops of the selected transport types.

Exceptions:

Type	Description
NotImplementedError	If the method is not implemented for the stop transport type in this city.

Source code in now8_api/service/city_data/__init__.py

@abstractmethod
async def get_stops_city(
    self,
    transport_types: List[TransportType] = None,
) -> List[Stop]:
    """Return all the stops of the selected transport types.

    Arguments:
        transport_types: Transport types to get the stops for.
            If none are passed, all of the available ones for the
            city will be included.

    Returns:
        All the stops of the selected transport types.

    Raises:
        NotImplementedError: If the method is not implemented for
            the stop transport type in this city.
    """

get_stops_route(self, route) async

Return all the stops of the selected route.

Parameters:

Name	Type	Description	Default
route	Route	Transport route to get the stops for.	required

Returns:

Type	Description
Tuple[List[now8_api.domain.Stop], List[now8_api.domain.Stop]]	Stops of the selected route for both ways in order.

Exceptions:

Type	Description
NotImplementedError	If the method is not implemented for the route transport type in this city.

Source code in now8_api/service/city_data/__init__.py

@abstractmethod
async def get_stops_route(
    self,
    route: Route,
) -> Tuple[List[Stop], List[Stop]]:
    """Return all the stops of the selected route.

    Arguments:
        route: Transport route to get the stops for.

    Returns:
        Stops of the selected route for both ways in order.

    Raises:
        NotImplementedError: If the method is not implemented for
            the route transport type in this city.
    """

UpstreamError (Exception)

Custom exception for upstream API fauilures.

Source code in now8_api/service/city_data/__init__.py

class UpstreamError(Exception):
    """Custom exception for upstream API fauilures."""

    def __init__(self, message):
        """Initialization method.

        Arguments:
            message: Error message.
        """
        super().__init__(message)

__init__(self, message) special

Initialization method.

Parameters:

Name	Type	Description	Default
message		Error message.	required

Source code in now8_api/service/city_data/__init__.py

def __init__(self, message):
    """Initialization method.

    Arguments:
        message: Error message.
    """
    super().__init__(message)

get_json(url) async

Fetch the given URL and returns the result as a dictionary.

Arguments: url: URL to fetch.

Returns:

Type	Description
dict	List of dictionaries with the parsed answer.

Source code in now8_api/service/city_data/__init__.py

@retry(
    retry=(
        retry_if_exception_type(TimeoutError)
        | retry_if_exception_type(aiohttp.ClientResponseError)
    ),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=0.5),
)
@validate_arguments
async def get_json(url: HttpUrl) -> dict:
    """Fetch the given URL and returns the result as a dictionary.

     Arguments:
        url: URL to fetch.

    Returns:
        List of dictionaries with the parsed answer.
    """
    async with aiohttp.ClientSession() as session:
        async with session.get(
            url,
            raise_for_status=True,
            timeout=10,
        ) as resp:
            return await resp.json(content_type=None)

---

Last update: 2022-08-05