San Francisco MTC
Fetching data from San Francisco MTC with Airbyte and Python.
Description
This Airbyte Connector uses the HTTP API provided by the San Francisco MTC in Python. With this connector, data is easily pulled from San Francisco MTC and loaded into the ViaWarn Postgres database, and are available in the ViaWarn app.
The connector is built on top of the Airbyte open-source platform, which provides a standardized way of integrating data sources and destinations. It is designed to be simple and easy to use. The required configuration is show in the the sections below.
The pulled resources from the AFMTA API are the following:
Requirements and Installation
Customized files
After creating the Airbyte connector template, modify the next files to retrieve data from San Francisco MTC.
spec.yaml
In this case, the connector password and the request limit are specified in the spec file.
documentationUrl: https://511.org/about/faq/open-data
connectionSpecification: $schema:http://json-schema.org/draft-07/schema#
title: VIAWARN US SF MTC Spec
type: object
required:
-developer_api_key
-request_limit
properties:
developer_api_key:
type: string
airbyte_secret: true
description: Developer API Key obtained from https://511.org/open-data/token
request_limit:
type: integer
description: Request limit as described here https://511.org/sites/default/f
config.json
Content:
{
"developer_api_key":"[key]",
"request_limit":"[limit(INTEGER)]"
}
configure_catalog.json
{
"streams":
[
{
"stream": {
"name": [STREAM_NAME],
"json_schema": {},
"supported_sync_modes": ["full_refresh"]
},
"sync_mode": "full_refresh", "destination_sync_mode": "overwrite"
}
]
}
source.py
Location: source-viawarn-us-sf-mtc/source.py
class ViawarnUsSfMtcStream()
- url_base
This field is the common part of the URL for all streams, if it does not exist, add it in each stream method.
String
- __init__()
This is a constructor method used to initialize the class. It uses the ‘config’ object to define the ‘developer_api_key’ and ‘request_limit’ attributes.
return config[]
- next_page_token()
If pagination is required, the response json would have two more elements: ‘pagination’ and ‘next_url’. Use them to set the ‘offset’ parameter.
return dictionary if there pagination is required, or None otherwise.
- request_params()
Create a dictionary. Keys:
‘developer_api_key’ and ‘limit’= request_limit. If pagination is available add ‘next_page_token’ key.
return a dictionary
- parse_response()
It is a placeholder to store the request.response object and constructs the appropriate mapping object for each parsed element by each Stream class.
yields an empty dictionary.
- backoff_time()
Time for a next tried in seconds. This is needed because the API states a request-rate limit of 60 requests per 3600s.
return number
class EVENT()
- primary_key
Unique field required in the stream associate to each output record.
String
- path()
subdomain and paramters to pass into the url: traffic/events
return a string
- parse_response()
Definition of how a response is parsed.
Set a definition to: primary key and ewkt fields. The response json is decoded using ‘ response.encoding=”‘utf-8-sig'” ‘
return an iterable containing each record in the response.
class SourceViawarnSfMtc()
- check_connection()
Set up a test to verify the connection between the connector and the API. This test should return two variables: the first one is a boolean that indicates the status of the connection, while the second one shows a log message if the connection fails.
return True, None
- streams()
