otb-kbo-webservice-pyclient

Downloads enterprise data from the Belgian KBO/BCE public webservice (kbopub) and stores one XML file per enterprise. Provides the kbo-fetch command line tool and a library API for use from other scripts.

It calls the ReadEnterprise SOAP operation and stores the ReadEnterpriseReply element of each reply, passed through as the service sent it apart from indentation and the namespace prefixes. The datamodel namespace is written with the ns2 prefix, which readers of these files match as literal text, so it is fixed rather than a matter of taste.

The webservice is spoken over the standard library, with no SOAP framework involved. boto3 is the one third-party dependency, used for S3 output and Secrets Manager credentials and imported only when one of those is reached.

Installation

uv sync              # library and CLI
uv sync --extra dev  # adds pytest, coverage, mypy and ruff

To use it from another project, install it from a checkout or from the repository:

pip install /path/to/otb-kbo-webservice-pyclient

Authentication

No secret name is built into the package; the installation names its own. Credentials are looked up in this order, the first match winning:

What a caller passes therefore beats whatever the process happens to have in its environment, and within each pair a username and password beat a secret to go and read. Naming a secret and setting the username and password is a normal state rather than a conflict: the credentials are used and the secret is left unread.

A secret is an AWS Secrets Manager secret whose string is JSON with user and password fields, and reading one needs AWS credentials in the environment. With nothing configured at all the run stops immediately, naming every option, without attempting an AWS call.

From the command line, set the environment:

export KBO_WEBSERVICE_USER=… KBO_WEBSERVICE_PASSWORD=…   # or
export KBO_WEBSERVICE_SECRET_NAME=my-org/kbo/webservice

From Python, pass what you have:

from kbo_webservice_client import resolve_credentials

resolve_credentials(secret_id="my-org/kbo/webservice")
resolve_credentials(username="…", password="…")
resolve_credentials()  # falls back to the environment

Passing only one of username and password is an error, since it can only be a mistake. A half-set environment is ignored instead, falling through to the next source, because it may not be the caller's doing.

Passwords never appear in log output.

Command line

# one number, written to the current directory
kbo-fetch --input BE0314595348

# a few numbers by hand: repeat the flag
kbo-fetch -i BE0314595348 -i 0403199702 -i 1234567894

# a list of numbers, written to a directory
kbo-fetch --input-file vats.txt --output-dir ./results

# straight to S3, with progress logged to stderr
kbo-fetch -f vats.txt -o s3://my-bucket/kbo/replies -v

# the list itself read from S3, so a job that produced it need not stage it
kbo-fetch -f s3://my-bucket/kbo/vats.txt -o s3://my-bucket/kbo/replies

# full debug output into a log file
kbo-fetch -f vats.txt -o ./results -vv --log-file run.log

Input may be enterprise numbers or VAT numbers, in any written form: 0314595348, BE0314595348 and BE 0314.595.348 are the same enterprise. An input file holds one number per line; blank lines and lines starting with # are ignored, and duplicates are fetched only once.

Give -i once per number to fetch a handful by hand; a comma-separated list in one value is not accepted, since a comma is never part of a number. For longer lists use --input-file, which may be a local path or an s3://bucket/key URL naming a single object, the same choice --output-dir offers. An input that cannot be read at all is fatal either way, since it leaves nothing to process.

Output files are named after the VAT form and the current date, whichever form the input took: BE0314595348_20260730.xml. Each holds the ReadEnterpriseReply element of the reply, indented.

Options

One of --input or --input-file is required. Exit codes:

Library

from kbo_webservice_client import KboClient, read_numbers_from_file, resolve_credentials

client = KboClient(resolve_credentials(), "s3://my-bucket/kbo/replies")
batch = read_numbers_from_file("s3://my-bucket/kbo/vats.txt")  # or a local path

for rejected in batch.rejected:
    print(f"skipping {rejected.value}: {rejected.rejection.value}")

report = client.fetch_all(batch.numbers)
print(report.summary())  # "12 written, 0 skipped, 1 not found, 0 failed, 4699 credits left"
print(report.credits_left)

Fetching a single enterprise:

from kbo_webservice_client import FetchOutcome, KboClient, parse_number, resolve_credentials

client = KboClient(resolve_credentials(), "./results")
result = client.fetch(parse_number("BE0314595348"))

if result.outcome is FetchOutcome.WRITTEN:
    print(result.location, result.reply.snapshot_date)

The output destination must be given explicitly when using the library; only the command line falls back to the working directory. It accepts a path, an s3:// URL, or any object with exists and write methods.

Validating numbers without fetching anything:

from kbo_webservice_client import InvalidNumberError, parse_number

try:
    number = parse_number("BE 0314.595.349")
except InvalidNumberError as error:
    print(error.rejection.value)  # "the check digits do not match the rest of the number"

Logging follows the standard library. The package logs under the kbo_webservice_client logger and configures nothing on import, so an application keeps control of its own handlers. configure_logging is available if the command line's setup is wanted.

Behaviour worth knowing

• Malformed input never reaches the service. A number is checked for length, an 0/1 prefix and its mod-97 check digits first. Rejected entries are logged with the original value and skipped, so no credit is spent on them.
• A number unknown to KBO is not a failure. The call is made, the status code is logged with the input value, and no file is written.
• Replies already stored today are skipped, which makes a re-run cheap and resumable. Use --no-skip-existing to force a refetch.
• A run stops early when the account runs out of credits, when the output destination refuses a write, or when the service rejects the connection with a SOAP fault. Everything else is logged and the run continues with the next number.
• More than 3000 numbers in one run logs a warning, in case the input file is not what was intended.
• Numbers are fetched sequentially with a short delay, as the service is metered per call.
• s3:// is recognised by its scheme, for both input and output. Anything else is a local path, and boto3 is only imported once an S3 location is actually reached.

Development

uv sync --extra dev
uv run pytest                       # the test suite
uv run coverage run -m pytest       # with coverage
uv run coverage report
uv run mypy                         # strict type checking
uv run ruff check . && uv run ruff format --check .

The test suite makes no network calls and needs no AWS account. External boundaries are stubbed: a scripted transport stands in for the webservice, an in-memory object store for S3, and recorded XML replies live in tests/fixtures/. The two functions that construct boto3 clients are the only code excluded from coverage.