Fetching data from CBP with Airbyte and Python.

Country wide CBP

Fetching data from Country wide CBP with Airbyte and Python.

Description

This Airbyte Connector uses the HTTP API provided by the Country wide CBP in Python. With this connector, data is easily pulled from Country wide CBP 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 CBP API are the following:

Requirements and Installation

Find the requirements and steps to create the connector in the Airbyte documentation.

Customized files

After creating the Airbyte connector template, modify the next files to retrieve data from Country wide CBP.

spec.yaml

Location: source-viawarn-us-az-dot/spec.yaml

Access API: Free
spec.yaml:

				
					documentationUrl: https://bwt.cbp.gov/
connectionSpecification:
  $schema: http://json-schema.org/draft-07/schema#
  title: Viawarn USA CW CBP - Spec
  type: object
  properties:
    documentationUrl:
      type: string
      description: You can consult https://bwt.cbp.gov/Viewallports.html?com=1&pas=1&ped=1&plist=0708

config.json

Location: secrets/config.json

This file is ignore by git.

The config file holds an empty directory because this API is FREE and does not require any other parameter.

config.json:

{}

configure_catalog.json

Location: integration_tests/configure_catalog.json

List of Streams that describes the data in the data store that a Source represents. It also declares the sync modes supported by the stream (full refresh or incremental).

configure_catalog.json:

				
					{ 
"streams": 
    [
    {
        "stream": {
            "name": "delay",
            "json_schema": {},
            "supported_sync_modes": ["full_refresh"] 
            },
        "sync_mode": "full_refresh", "destination_sync_mode": "overwrite" 
        } 
    ]
}

source.py

Location: source-viawarn-us-cw-cbp/source.py

class ViawarnUsCwCbpStream()

Create a stream to represent the data that will be pull from the API. It contains all the common API functionalities:

url_base

This field is the first part of the URL for the streams.

String

next_page_token()

Pagination is not provided by this API.

return None

request_params()

The API does not require any parameter.

return a None

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.

class Delay()

This method extends the ViawarnUsCwCbpStream() method to specified unique behavior for this stream.

primary_key

Unique field required in the stream associate to each output record.

String

path()

subdomain and parameters to pass into the URL.

return a string

parse_response()

Definition of how a response is parsed.

Set a definition to: primary key.

return an iterable containing each record in the response.

class SourceViawarnUsCwCbp()

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()

Set up the stream in the source that will be parsed.

return List of streams

delay.json

Location: source-viawarn-us-cw-cbp/schemas/delay.json

Declare schemas for the stream by adding a json file describing the output from that stream.

Streams name: delay

Resources

LANCER L'APPLI

Support

support@viawarn.com