TransitLand

TransitLandDownloader

TransitLandDownloader(api_key: Optional[str] = None)

Bases: BaseGTFSDownloader

Client for searching and downloading feeds from Transitland.

See https://www.transit.land/documentation/rest-api for the full API reference. Requires a Transitland API key.

Initialize the client.

Parameters:

Name	Type	Description	Default
api_key	Optional[str]	Transitland API key. If not provided, it is read from the TRANSITLAND_API_KEY environment variable.	None

Raises:

Type	Description
ValueError	If no API key is available.

Source code in pyGTFSHandler/downloaders/transitland.py

def __init__(self, api_key: Optional[str] = None):
    """Initialize the client.

    Args:
        api_key: Transitland API key. If not provided, it is read
            from the `TRANSITLAND_API_KEY` environment variable.

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

search_feeds

search_feeds(aoi: Optional[AOIType] = None, lat: Optional[float] = None, lon: Optional[float] = None, radius: Optional[float] = None, country_code: Optional[str] = None, state: Optional[str] = None, city: Optional[str] = None, search: Optional[str] = None, operator_onestop_id: Optional[str] = None, limit: Optional[int] = 1000, spec: str = 'gtfs') -> List[GTFSFeedMetadata]

Search Transitland's catalog for feeds matching filters.

Parameters:

Name	Type	Description	Default
aoi	Optional[AOIType]	Polygon, MultiPolygon, GeoDataFrame, or GeoSeries defining a geographic area; converted to a bounding box (bbox query parameter). Mutually exclusive with lat/lon.	None
lat	Optional[float]	Latitude for a point-radius search. Requires lon.	None
lon	Optional[float]	Longitude for a point-radius search. Requires lat.	None
radius	Optional[float]	Search radius in meters around (lat, lon).	None
country_code	Optional[str]	ISO 3166-1 alpha-2 country code to filter by (Transitland's adm0_iso).	None
state	Optional[str]	State/province/subdivision name to filter by (Transitland's adm1_name).	None
city	Optional[str]	City name to filter by (Transitland's city_name).	None
search	Optional[str]	Free-text search over feed and operator names.	None
operator_onestop_id	Optional[str]	Restrict to feeds of a specific operator, identified by its Transitland onestop id.	None
limit	Optional[int]	Maximum number of feeds to return. Results are fetched page by page (up to MAX_PAGE_SIZE per request) following the API's cursor until this many feeds are collected or the catalog is exhausted.	1000
spec	str	GTFS specification to filter by. Defaults to "gtfs" (static GTFS); Transitland also catalogs other specs such as "gtfs-rt".	'gtfs'

Returns:

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

Raises:

Type	Description
ValueError	If only one of lat/lon is given.

Source code in pyGTFSHandler/downloaders/transitland.py

def search_feeds(
    self,
    aoi: Optional[AOIType] = None,
    lat: Optional[float] = None,
    lon: Optional[float] = None,
    radius: Optional[float] = None,
    country_code: Optional[str] = None,
    state: Optional[str] = None,
    city: Optional[str] = None,
    search: Optional[str] = None,
    operator_onestop_id: Optional[str] = None,
    limit: Optional[int] = 1000,
    spec: str = "gtfs",
) -> List[GTFSFeedMetadata]:
    """Search Transitland's catalog for feeds matching filters.

    Args:
        aoi: Polygon, MultiPolygon, GeoDataFrame, or GeoSeries defining
            a geographic area; converted to a bounding box (`bbox`
            query parameter). Mutually exclusive with `lat`/`lon`.
        lat: Latitude for a point-radius search. Requires `lon`.
        lon: Longitude for a point-radius search. Requires `lat`.
        radius: Search radius in meters around `(lat, lon)`.
        country_code: ISO 3166-1 alpha-2 country code to filter by
            (Transitland's `adm0_iso`).
        state: State/province/subdivision name to filter by
            (Transitland's `adm1_name`).
        city: City name to filter by (Transitland's `city_name`).
        search: Free-text search over feed and operator names.
        operator_onestop_id: Restrict to feeds of a specific operator,
            identified by its Transitland onestop id.
        limit: Maximum number of feeds to return. Results are fetched
            page by page (up to `MAX_PAGE_SIZE` per request) following
            the API's cursor until this many feeds are collected or
            the catalog is exhausted.
        spec: GTFS specification to filter by. Defaults to `"gtfs"`
            (static GTFS); Transitland also catalogs other specs such
            as `"gtfs-rt"`.

    Returns:
        A list of `GTFSFeedMetadata`, one per matching feed with a
        usable static GTFS download URL.

    Raises:
        ValueError: If only one of `lat`/`lon` is given.
    """
    if (lat is None) != (lon is None):
        raise ValueError("Both 'lat' and 'lon' must be provided together.")

    params: Dict[str, Any] = {
        "spec": spec,
        "search": search,
        "adm0_iso": country_code,
        "adm1_name": state,
        "city_name": city,
        "operator_onestop_id": operator_onestop_id,
    }

    if aoi is not None:
        min_lon, min_lat, max_lon, max_lat = bbox_from_aoi(aoi)
        params["bbox"] = f"{min_lon},{min_lat},{max_lon},{max_lat}"

    if lat is not None:
        params["lat"] = lat
        params["lon"] = lon
        if radius is not None:
            params["radius"] = radius

    raw_feeds = self._paginated_search(params, limit=limit)
    return [
        feed
        for feed in (self._to_feed_metadata(raw) for raw in raw_feeds)
        if feed is not None
    ]

find_feed_version_history

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

List every published static GTFS version recorded for a feed.

Uses GET /feeds/{feed_key}/feed_versions, which returns Transitland's full archive of past fetches for the feed, each tagged with earliest_calendar_date/latest_calendar_date -- the real service period Transitland computed from that version's own calendar.txt/calendar_dates.txt, not just when it was fetched (fetched_at).

Parameters:

Name	Type	Description	Default
feed_key	str	Feed lookup key: an integer id or onestop id.	required
limit	int	Maximum number of feed versions to fetch.	1000

Returns:

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

Source code in pyGTFSHandler/downloaders/transitland.py

def find_feed_version_history(self, feed_key: str, limit: int = 1000) -> List[Dict[str, Any]]:
    """List every published static GTFS version recorded for a feed.

    Uses `GET /feeds/{feed_key}/feed_versions`, which returns
    Transitland's full archive of past fetches for the feed, each
    tagged with `earliest_calendar_date`/`latest_calendar_date` -- the
    real service period Transitland computed from that version's own
    `calendar.txt`/`calendar_dates.txt`, not just when it was fetched
    (`fetched_at`).

    Args:
        feed_key: Feed lookup key: an integer id or onestop id.
        limit: Maximum number of feed versions to fetch.

    Returns:
        Raw feed_version dictionaries, sorted oldest to newest by
        `earliest_calendar_date` (entries missing that field sort
        last, in API order).
    """
    data = self._get(f"{self.BASE_URL}/feeds/{feed_key}/feed_versions", {"limit": limit})
    versions = data.get("feed_versions", [])
    versions.sort(key=lambda v: self._version_start(v) or datetime.max)
    return versions

download_historic_stack

download_historic_stack(output_path: str, feed_key: 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 version history into one GTFS zip.

Mirrors NAPDownloader.download_historic_stack/ MobilityDatabaseDownloader.download_historic_stack, using find_feed_version_history's earliest_calendar_date as the trim anchor -- already a real, Transitland-computed service start date, unlike NAP's raw publication timestamp.

Note: downloading any feed_version other than the single most recent one requires a paid (or free Hobbyist/Academic) Transitland plan; see _download_feed_version.

Parameters:

Name	Type	Description	Default
output_path	str	Directory to assemble the stitched feed in.	required
feed_key	str	Feed lookup key: an integer id or onestop id.	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_key}_{start}_{end}.zip,
Optional[str]	or None if no feed versions were found covering the
Optional[str]	requested range.

Source code in pyGTFSHandler/downloaders/transitland.py

def download_historic_stack(
    self,
    output_path: str,
    feed_key: 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 version history into one GTFS zip.

    Mirrors `NAPDownloader.download_historic_stack`/
    `MobilityDatabaseDownloader.download_historic_stack`, using
    `find_feed_version_history`'s `earliest_calendar_date` as the
    trim anchor -- already a real, Transitland-computed service start
    date, unlike NAP's raw publication timestamp.

    Note: downloading any feed_version other than the single most
    recent one requires a paid (or free Hobbyist/Academic) Transitland
    plan; see `_download_feed_version`.

    Args:
        output_path: Directory to assemble the stitched feed in.
        feed_key: Feed lookup key: an integer id or onestop id.
        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_key}_{start}_{end}.zip`,
        or `None` if no feed 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_key))
    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_feed_version_history(feed_key)
    versions = select_versions_covering_range(
        history, start_date, end_date, get_start=self._version_start
    )
    for version in versions:
        version["start_date"] = self._version_start(version)

    if not versions:
        logger.warning(f"No feed versions found for feed '{feed_key}' 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_feed_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 feed '{main_name}'.")

    cleanup_version_paths(path_stack)
    shutil.rmtree(main_path)

    return zip_path