speedracer 0.0.2

Conveniently access data

What is speedracer?

speedracer is an asyncio enabled python library for interacting with Autobahn.

• Subscribe to datasets available in the Mission Data Broker (MDB)
• Authenticate with Autobahn APIs

Quickstart

Subscribe to a dataset

To subscribe to a dataset on the MDB

1. Provide the base URL of the MDB Account Server (MAS) when creating a Connection
2. Provide the dataset to the subscribe method

conn = Connection('https://mdb-account-server/')
sub = await conn.subscribe('mydataset')
async for msg in sub:
    print(msg.data.decode())

Authenticate with Autobahn APIs

Autobahn uses JWT credentials for authorization. These tokens are put into the Authorization header for all HTTPs requests.

JWTs are obtained by exchanging private keys with the API Gateway server and must be periodically refreshed.

speedracer automates the process of fetching JWTs and refreshing them when they expire.

manager = JWTManager('https://api-gateway/getjwt')
auth_headers = manager.get_headers()
requests.get('https://autobahn-service/', headers=auth_headers)

Advanced Usage

Dataset Subscriptions

Callback

Instead of iterating over a Subscription, you can provide a callback that takes a message as an argument.

def cb(msg):
    print(msg.data.decode())

conn = Connection('https://mdb-account-server.com')
sub = await conn.subscribe('mydataset', callback=cb)
await sub.wait(messages=10) # exit after 10 messages received

Seek

Autobahn maintains a short history of messages (by default 7 days) for each dataset. To navigate to different points in the stream use the seek method. Seek accepts a message sequence number or a datetime object.

# seek to message sequence 1
await sub.seek(1)

# seek to 5 minutes ago
await sub.seek(datetime.datetime.utcnow() - datetime.timedelta(minutes=5))

Offset

By default, subscribe starts at the current time; to start at the beginning of the stream use Offset.BEGIN

sub = await conn.subscribe('mydataset', offset=Offset.BEGIN)

Low Latency

By default, each Subscription fetches messages in batches of 10. If the server has less than 10 messages to send, it will wait up to 10 seconds for new messages before returning the messages it does have. For low-latency applications set the batch size to 1.

sub = await conn.subscribe('mydataset', batch=1)

Important Notes

TLS Errors

TLS errors occur for 2 reasons

1. Invalid or missing client certificates
2. No certificate authority configured

Client certificates

By default, speedracer uses PEM formatted X.509 certificates for user authentication. By default speedracer will use certificates placed in the users $HOME/.ssh directory. The certificate and key files should be named $HOME/.ssh/cert.crt and $HOME/.ssh/cert.key, respectively.

To specify a different path, pass a tuple containing the path to the certificate and key files to the cert argument.

conn = Connection('https://mdb-account-server', cert=('./mycert.crt', './mycert.key'))

Certificate Authority

Certificate Authorities establish a chain of trust. This is usually configured globally as part of the operating system. There are cases where the OS does not have the proper certificate authority configured.

In these instances, users may specify a cabundle as a X.509 PEM file via the environment variable SSL_CA_BUNDLE or via the arugment ca_bundle.

conn = Connection('https://mdb-account-server.com', ca_bundle='ca.pem')

Subscription not starting at specified offset

Subscriptions to datasets are durable. This means that once you have subscribed to a dataset, message delivery restarts where you left off--even if your program restarts.

Subscriptions will expire after 1-week if not used.