Mobility Database

MobilityDatabaseDownloader

MobilityDatabaseDownloader(api_key: Optional[str] = None)

Bases: BaseGTFSDownloader

Client for searching and downloading feeds from the Mobility Database.

Authenticates via OAuth2 using a long-lived refresh token (obtained from the Mobility Database website), which is exchanged for short-lived access tokens as needed.

Initialize the client and obtain an initial access token.

Parameters:

Name	Type	Description	Default
api_key	Optional[str]	Long-lived refresh token obtained from the Mobility Database website. If not provided, it is read from the MOBILITY_DATABASE_REFRESH_TOKEN environment variable.	None

Raises:

Type	Description
ValueError	If no refresh token is available.

Source code in pyGTFSHandler/downloaders/mobility_database.py

def __init__(self, api_key: Optional[str] = None):
    """Initialize the client and obtain an initial access token.

    Args:
        api_key: Long-lived refresh token obtained from the Mobility
            Database website. If not provided, it is read from the
            `MOBILITY_DATABASE_REFRESH_TOKEN` environment variable.

    Raises:
        ValueError: If no refresh token is available.
    """
    super().__init__(api_key=api_key)
    if not self.api_key:
        raise ValueError(
            "A refresh token is required (pass api_key or set "
            f"{self.API_KEY_ENV_VAR})."
        )

    self._access_token: Optional[str] = None
    self._token_expires_at: Optional[datetime] = None
    self._ensure_access_token()

search_feeds

search_feeds(aoi: Optional[AOIType] = None, limit: Optional[int] = 100000, offset: Optional[int] = None, provider: Optional[Union[str, List[Optional[str]]]] = None, producer_url: Optional[Union[str, List[Optional[str]]]] = None, country_code: Optional[Union[str, List[Optional[str]]]] = None, subdivision_name: Optional[Union[str, List[Optional[str]]]] = None, municipality: Optional[Union[str, List[Optional[str]]]] = None, bounding_filter_method: str = 'partially_enclosed', is_official: Optional[bool] = None) -> List[GTFSFeedMetadata]

Search the Mobility Database for GTFS feeds matching filters.

Supports multiple values per filter (client-side merged across combinations) and automatic pagination for limit values above the API's per-request page size (200).

Parameters:

Name	Type	Description	Default
aoi	Optional[AOIType]	Polygon, MultiPolygon, GeoDataFrame, or GeoSeries defining a geographic area to filter by.	None
limit	Optional[int]	Maximum number of feeds to return. Values above 200 trigger automatic multi-page fetching.	100000
offset	Optional[int]	Offset for pagination.	None
provider	Optional[Union[str, List[Optional[str]]]]	Provider name filter. Accepts a string, a list of strings, or a list including None to also match feeds with no provider set.	None
producer_url	Optional[Union[str, List[Optional[str]]]]	Producer URL filter, same semantics as provider.	None
country_code	Optional[Union[str, List[Optional[str]]]]	Country code filter, same semantics as provider.	None
subdivision_name	Optional[Union[str, List[Optional[str]]]]	Subdivision name filter, same semantics as provider.	None
municipality	Optional[Union[str, List[Optional[str]]]]	Municipality filter, same semantics as provider.	None
bounding_filter_method	str	Spatial inclusion rule when aoi is set. One of "completely_enclosed", "partially_enclosed", "disjoint".	'partially_enclosed'
is_official	Optional[bool]	If set, filter by whether the feed is official.	None

Returns:

Type	Description
List[GTFSFeedMetadata]	A list of GTFSFeedMetadata, one per matching feed with a
List[GTFSFeedMetadata]	usable download URL.

Source code in pyGTFSHandler/downloaders/mobility_database.py

def search_feeds(
    self,
    aoi: Optional[AOIType] = None,
    limit: Optional[int] = 100_000,
    offset: Optional[int] = None,
    provider: Optional[Union[str, List[Optional[str]]]] = None,
    producer_url: Optional[Union[str, List[Optional[str]]]] = None,
    country_code: Optional[Union[str, List[Optional[str]]]] = None,
    subdivision_name: Optional[Union[str, List[Optional[str]]]] = None,
    municipality: Optional[Union[str, List[Optional[str]]]] = None,
    bounding_filter_method: str = "partially_enclosed",
    is_official: Optional[bool] = None,
) -> List[GTFSFeedMetadata]:
    """Search the Mobility Database for GTFS feeds matching filters.

    Supports multiple values per filter (client-side merged across
    combinations) and automatic pagination for `limit` values above
    the API's per-request page size (200).

    Args:
        aoi: Polygon, MultiPolygon, GeoDataFrame, or GeoSeries defining
            a geographic area to filter by.
        limit: Maximum number of feeds to return. Values above 200
            trigger automatic multi-page fetching.
        offset: Offset for pagination.
        provider: Provider name filter. Accepts a string, a list of
            strings, or a list including `None` to also match feeds
            with no provider set.
        producer_url: Producer URL filter, same semantics as `provider`.
        country_code: Country code filter, same semantics as `provider`.
        subdivision_name: Subdivision name filter, same semantics as
            `provider`.
        municipality: Municipality filter, same semantics as `provider`.
        bounding_filter_method: Spatial inclusion rule when `aoi` is
            set. One of "completely_enclosed", "partially_enclosed",
            "disjoint".
        is_official: If set, filter by whether the feed is official.

    Returns:
        A list of `GTFSFeedMetadata`, one per matching feed with a
        usable download URL.
    """
    MAX_PAGE = 200
    kwargs = dict(
        aoi=aoi,
        provider=provider,
        producer_url=producer_url,
        country_code=country_code,
        subdivision_name=subdivision_name,
        municipality=municipality,
        bounding_filter_method=bounding_filter_method,
        is_official=is_official,
    )

    if limit is None:
        limit = MAX_PAGE

    if limit <= MAX_PAGE:
        raw_feeds = self._search_feeds_page(limit=limit, offset=offset, **kwargs)
    else:
        raw_feeds = []
        used_ids = set()
        remaining = limit
        current_offset = offset if offset is not None else 0

        while remaining > 0:
            page_limit = min(MAX_PAGE, remaining)
            page_results = self._search_feeds_page(
                limit=page_limit, offset=current_offset, **kwargs
            )
            if not page_results:
                break

            for feed in page_results:
                fid = feed.get("id")
                if fid and fid not in used_ids:
                    raw_feeds.append(feed)
                    used_ids.add(fid)

            remaining -= page_limit
            current_offset += page_limit
            if len(page_results) < page_limit:
                break

    return [
        feed
        for feed in (self._to_feed_metadata(raw) for raw in raw_feeds)
        if feed is not None
    ]

find_dataset_history

find_dataset_history(feed_id: str, limit: int = 1000) -> List[Dict[str, Any]]

List every published dataset (version) recorded for a feed.

Uses /v1/gtfs_feeds/{id}/datasets, which -- unlike search_feeds's latest_dataset -- returns the full publication history MobilityData has recorded for the feed, each already tagged with service_date_range_start/service_date_range_end: the real service period MobilityData computed from that dataset's own calendar.txt/calendar_dates.txt, not just when it was downloaded.

Parameters:

Name	Type	Description	Default
feed_id	str	Mobility Database feed id (e.g. "mdb-10").	required
limit	int	Maximum number of dataset versions to fetch.	1000

Returns:

Type	Description
List[Dict[str, Any]]	Raw dataset dictionaries, sorted oldest to newest by
List[Dict[str, Any]]	service_date_range_start (entries missing that field sort
List[Dict[str, Any]]	last, in API order).

Source code in pyGTFSHandler/downloaders/mobility_database.py

def find_dataset_history(
    self,
    feed_id: str,
    limit: int = 1000,
) -> List[Dict[str, Any]]:
    """List every published dataset (version) recorded for a feed.

    Uses `/v1/gtfs_feeds/{id}/datasets`, which -- unlike
    `search_feeds`'s `latest_dataset` -- returns the full publication
    history MobilityData has recorded for the feed, each already
    tagged with `service_date_range_start`/`service_date_range_end`:
    the real service period MobilityData computed from that dataset's
    own `calendar.txt`/`calendar_dates.txt`, not just when it was
    downloaded.

    Args:
        feed_id: Mobility Database feed id (e.g. `"mdb-10"`).
        limit: Maximum number of dataset versions to fetch.

    Returns:
        Raw dataset dictionaries, sorted oldest to newest by
        `service_date_range_start` (entries missing that field sort
        last, in API order).
    """
    datasets = self._authorized_request(
        "GET", f"{self.GTFS_FEEDS_ENDPOINT}/{feed_id}/datasets", params={"limit": limit}
    )
    datasets.sort(key=lambda d: self._dataset_start(d) or datetime.max)
    return datasets

download_historic_stack

download_historic_stack(output_path: str, feed_id: str, start_date: DateLike, end_date: DateLike, day_separation: int = 1, overwrite: bool = False, aoi: Optional[AOIType] = None) -> Optional[str]

Download and stitch a feed's dataset history into one GTFS zip.

Mirrors NAPDownloader.download_historic_stack, but since find_dataset_history already reports each version's real service date range (computed by MobilityData from that version's own calendar), no extra validation of the anchor date is needed before trimming (contrast with NAP's raw publication timestamp, handled by downloaders.spain.utils.resolve_publication_start_date).

Parameters:

Name	Type	Description	Default
output_path	str	Directory to assemble the stitched feed in.	required
feed_id	str	Mobility Database feed id (e.g. "mdb-10").	required
start_date	DateLike	Start of the date range to cover. Accepts "today", a date/datetime, or an ISO "YYYY-MM-DD" string.	required
end_date	DateLike	End of the date range to cover.	required
day_separation	int	Minimum number of days a version is assumed to stay valid for, if service periods don't force it shorter.	1
overwrite	bool	If True, redo the feed even if its stitched output zip already exists.	False
aoi	Optional[AOIType]	Optional AOI passed through to historic_stack to restrict stops.	None

Returns:

Type	Description
Optional[str]	Path to the written {SOURCE_NAME}_{feed_id}_{start}_{end}.zip,
Optional[str]	or None if no dataset versions were found covering the
Optional[str]	requested range.

Source code in pyGTFSHandler/downloaders/mobility_database.py

def download_historic_stack(
    self,
    output_path: str,
    feed_id: str,
    start_date: DateLike,
    end_date: DateLike,
    day_separation: int = 1,
    overwrite: bool = False,
    aoi: Optional[AOIType] = None,
) -> Optional[str]:
    """Download and stitch a feed's dataset history into one GTFS zip.

    Mirrors `NAPDownloader.download_historic_stack`, but since
    `find_dataset_history` already reports each version's real
    service date range (computed by MobilityData from that version's
    own calendar), no extra validation of the anchor date is needed
    before trimming (contrast with NAP's raw publication timestamp,
    handled by `downloaders.spain.utils.resolve_publication_start_date`).

    Args:
        output_path: Directory to assemble the stitched feed in.
        feed_id: Mobility Database feed id (e.g. `"mdb-10"`).
        start_date: Start of the date range to cover. Accepts
            `"today"`, a `date`/`datetime`, or an ISO `"YYYY-MM-DD"`
            string.
        end_date: End of the date range to cover.
        day_separation: Minimum number of days a version is assumed
            to stay valid for, if service periods don't force it
            shorter.
        overwrite: If True, redo the feed even if its stitched output
            zip already exists.
        aoi: Optional AOI passed through to `historic_stack` to
            restrict stops.

    Returns:
        Path to the written `{SOURCE_NAME}_{feed_id}_{start}_{end}.zip`,
        or `None` if no dataset versions were found covering the
        requested range.
    """
    os.makedirs(output_path, exist_ok=True)
    start_date, end_date = normalize_date_range(start_date, end_date)
    main_name = sanitize_filename(str(feed_id))
    zip_source_name = f"{self.SOURCE_NAME}_{main_name}"

    final_zip = os.path.join(
        output_path,
        f"{zip_source_name}_{start_date.strftime('%Y%m%d')}_{end_date.strftime('%Y%m%d')}.zip",
    )
    if not overwrite and os.path.isfile(final_zip):
        logger.info(f"'{final_zip}' already exists. Skipping.")
        return final_zip

    history = self.find_dataset_history(feed_id)
    versions = select_versions_covering_range(
        history, start_date, end_date, get_start=self._dataset_start
    )
    for version in versions:
        version["start_date"] = self._dataset_start(version)

    if not versions:
        logger.warning(f"No dataset versions found for feed '{feed_id}' in the requested range.")
        return None

    main_path = os.path.normpath(os.path.join(output_path, main_name))
    path_stack = download_and_stitch_versions(
        versions, main_path, day_separation, end_date, overwrite, self._download_dataset_version
    )
    if not path_stack:
        return None

    if os.path.isfile(path_stack[-1] + ".zip"):
        os.remove(path_stack[-1] + ".zip")

    historic_stack(path_stack, main_path, aoi)
    zip_path = zip_stitched_feed(main_path, zip_source_name, start_date, end_date, output_path)
    logger.info(f"Finished stitching historic dataset '{main_name}'.")

    cleanup_version_paths(path_stack)
    shutil.rmtree(main_path)

    return zip_path