How to access CH Streaming API via Python?

API Question
#1

I am struggling to figure out how to connect to the CH Streaming API via Python (requests library). I went through the documentation but was not able to solve my problem with it. I was able to connect to the REST API without any issues, but not to the Streaming API :frowning:

Please note that I removed my_api_key from the code because I want to keep my key save. I did take the stream key and not the REST API one.

import requests
url = 'https://stream.companieshouse.gov.uk/insolvency-cases'
headers = {'my_api_key': '***********my_api_key**********'}
r = requests.get(url, headers=headers)
print(r.text)

This results in the following error:
{“error”:“Empty Authorization header”,“type”:“ch:service”}

What am I doing wrong? Any pointer in the right direction will help. Thanks in advance.

#2

I’m not a Python coder but I suspect that you’re not using the correct headers there. That doesn’t look like the correct ones for http basic authentication

From the Python manuals online - how to do http basic authentication in Python:
https://docs.python-requests.org/en/latest/user/authentication/#basic-authentication

The lines that relate to this:

In fact, HTTP Basic Auth is so common that Requests provides a handy shorthand for using it:

requests.get(‘https://httpbin.org/basic-auth/user/pass’, auth=(‘user’, ‘pass’))

Don’t forget that for Companies House APIs your API key is the “user” and “pass” is an empty string. As noted everywhere - including the Companies House manual below - you can and probably should try doing this with e.g. Curl to ensure that your API key is correct / you’re running this somewhere that can reach the Companies House servers and that their servers recognise.

The key document from Companies House you need is this one I think:

https://developer-specs.company-information.service.gov.uk/streaming-api/guides/authentication

Good luck.

#3

This should work as a start unless I’ve mangled it in the copy. Filings or companies are better for testing as insolvency cases are less frequent:

import requests
api_key = '----- my api key -----'
url = 'https://stream.companieshouse.gov.uk/filings'
stream_timeout = 600   # time out ten mins just as example, no need if not wanted
"""
begin streaming, can drop the timeout if not required, maybe use try: except: too
"""
print('Streaming URL', url)
r = requests.get(url, auth=(api_key, ''), stream=True, timeout=stream_timeout)
print(r.status_code, r.headers)
"""
check for good requests status (200 = ok) and process the stream
"""
if r.status_code == 200:
    for json_line in r.iter_lines():     # stream should give a continuous iterable
        if json_line:
            print(json_line)
        else:
            print('Empty pulse')

else:
    input('Not 200, best check what to do')
1 Like