Canadian sanctions API in Python: autonomous list customer screening

{
  "subject": {
    "name": "Mira Calder",
    "entityType": "person",
    "birthDate": "1984"
  },
  "package": "sandbox@1",
  "retention": "standard"
}

"""Python 3.11+. Run on your trusted backend; standard library only."""
import json, os, sys
from pathlib import Path
from urllib.error import HTTPError
from urllib.request import Request, build_opener, HTTPRedirectHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl): return None

class ApiFailure(Exception): pass

def main():
    key, operation = os.environ['SANCTIONSKIT_API_KEY'], os.environ['SCREENING_OPERATION_ID']
    request = json.loads(Path(sys.argv[1] if len(sys.argv) > 1 else 'request.json').read_text(encoding='utf-8'))
    sandbox = request.get('package') == 'sandbox@1'
    source_ids = ['sandbox-synthetic'] if sandbox else request.get('sources')
    if not isinstance(source_ids, list) or not source_ids: raise ValueError('Select your sources.')
    opener = build_opener(NoRedirect())
    def api(path, body=None):
        headers = {'Authorization': 'Bearer ' + key}
        if body is not None:
            headers.update({'Content-Type': 'application/json', 'Idempotency-Key': operation})
        call = Request('https://www.sanctionskit.com/api/v1' + path, headers=headers,
                       data=None if body is None else json.dumps(body).encode(),
                       method='GET' if body is None else 'POST')
        try:
            response = opener.open(call, timeout=25)
        except HTTPError as error:
            response = error
        with response:
            envelope = json.load(response)
        if not 200 <= response.status < 300:
            code = envelope.get('error', {}).get('code', 'unknown')
            if not isinstance(code, str) or not 1 <= len(code) <= 80 or any(c not in 'abcdefghijklmnopqrstuvwxyz0123456789_' for c in code): code = 'unknown'
            raise ApiFailure('API HTTP ' + str(response.status) + ': ' + code)
        return envelope.get('data')
    if not sandbox:
        sources = api('/sources')
        if not isinstance(sources, list): raise ValueError('Invalid discovery response.')
        for source_id in source_ids:
            found = [source for source in sources if isinstance(source, dict) and source.get('id') == source_id]
            if len(found) != 1: raise ValueError('Required source missing or ambiguous.')
            source = found[0]
            if (source.get('availability') != 'available' or source.get('fresh') is not True or
                    not isinstance(source.get('capabilities'), list) or
                    request['subject']['entityType'] not in source['capabilities']):
                raise ValueError('Required source unavailable or incompatible.')
    result = api('/screenings', request)
    matches, coverage = result.get('matches'), result.get('coverage')
    if (not isinstance(result.get('id'), str) or not result['id'] or
            result.get('environment') != ('sandbox' if sandbox else 'production') or
            result.get('status') not in ['potential_match', 'no_match'] or
            not isinstance(matches, list) or (result['status'] == 'no_match') != (len(matches) == 0) or
            not isinstance(coverage, list) or len(coverage) != len(source_ids) or
            any(len([item for item in coverage if item.get('sourceId') == source_id and item.get('fresh') is True]) != 1 for source_id in source_ids)):
        raise ValueError('Incomplete screening result.')
    # Store result['id'] with your customer reference; retrieve evidence with results:read.
    print(json.dumps({'id': result['id'], 'status': result['status'], 'candidateCount': len(matches),
                      'sources': [item['sourceId'] for item in coverage]}))

try:
    main()
except ApiFailure as error:
    print(str(error), file=sys.stderr)
    sys.exit(1)
except Exception as error:
    print(str(error) if str(error) == 'Required source unavailable or incompatible.' else
          'Screening incomplete. Check credentials, request and source coverage.', file=sys.stderr)
    sys.exit(1)

python screen.py request.json

curl --fail-with-body --max-time 25 --header "Authorization: Bearer $SANCTIONSKIT_API_KEY" "https://www.sanctionskit.com/api/v1/sources"

{
  "subject": {
    "name": "Mira Calder",
    "entityType": "person",
    "birthDate": "1984"
  },
  "sources": [
    "ca-autonomous"
  ],
  "retention": "standard"
}