peppol_lookup 0.2.0

Python Client performing entity lookup with peppol

Peppol Lookup

This is client library for fetching business entities and verifying whether they support electronic invoices

Default api url should be: https://directory.peppol.eu

Table of contents

• Installation
• Usage
  • Api Usage
  • Retry strategies
    • Simple retry strategy
    • Progressive retry strategy
    • Writing custom retry strategy
• API methods
• Types
• Development
  • Testing
• Contributing

Installation

To install this package you can use following command

pip install peppol_lookup

Usage

Api usage

This client supports both sync and async clients depending on the use case. These can be also used with python context managers.

The api_url parameter is optional and defaults to https://directory.peppol.eu.

Sync version

# Using default URL
from peppol_lookup import PeppolApi

api = PeppolApi()

response = api.search("My Company")
print(response.data)

# With custom URL
from peppol_lookup import PeppolApi

api = PeppolApi(api_url="https://custom-api.example.com")

response = api.search("My Company")
print(response.data)

# Context manager
from peppol_lookup import PeppolApi

with PeppolApi() as client:
    response = client.search("My Company")

print(response.data)

Async version

# Using default URL
from peppol_lookup import AsyncPeppolApi

api = AsyncPeppolApi()

response = await api.search("My Company")
print(response.data)

# With custom URL
from peppol_lookup import AsyncPeppolApi

api = AsyncPeppolApi(api_url="https://custom-api.example.com")

response = await api.search("My Company")
print(response.data)

# Context manager
from peppol_lookup import AsyncPeppolApi

async with AsyncPeppolApi() as client:
    response = await client.search("My Company")

print(response.data)

Retry strategies

This API library supports 2 retry strategies for both syns and async clients. Simple and progressive retry strategies.

Simple retry strategy

This is basic retry strategy that tries request and if there is an error will wait constant time and try again. This process will repeat number of times that is defined

Example of usage:

# Example of simple retry strategy that will try 3 times with 1 second in-between

# Sync example
from peppol_lookup import PeppolApi, SimpleRetryStrategy
api = PeppolApi(api_url="https://custom-api.example.com", retry_strategy=SimpleRetryStrategy(retries=3, delay=1))
response = api.search("My Company")

# Async example
from peppol_lookup import AsyncPeppolApi, AsyncSimpleRetryStrategy
api = AsyncPeppolApi(api_url="https://custom-api.example.com", retry_strategy=AsyncSimpleRetryStrategy(retries=3, delay=1))
response = await api.search("My Company")

Progressive retry strategy

This strategy will have number of retries with progressive back-off by fixed factor.

Be careful when using this strategy since in sync mode can block execution for a long time

Example of usage:

# Example of progressive retry strategy that will try 3 times with delay multiplied
# by factor on every retry. In example bellow after first failure it will wait
# 1 second, then 3 seconds and finally 9 seconds

# Sync example
from peppol_lookup import PeppolApi, ProgressiveRetryStrategy
api = PeppolApi(api_url="https://custom-api.example.com", retry_strategy=ProgressiveRetryStrategy(retries=3, delay=1, factor=3))
response = api.search("My Company")

# Async example
from peppol_lookup import AsyncPeppolApi, AsyncProgressiveRetryStrategy
api = AsyncPeppolApi(api_url="https://custom-api.example.com", retry_strategy=AsyncProgressiveRetryStrategy(retries=3, delay=1, factor=3))
response = await api.search("My Company")

Writing custom retry strategy

If you need special strategy it is possible to create own strategies. You would need to extend one of the base classes depending whether you need Sync or Async version.

For example:

import httpx
from peppol_lookup import BaseRetryStrategy, AsyncBaseRetryStrategy, MethodData

class MyRetryStrategy(BaseRetryStrategy):
    def __init__(self, custom_property1, custom_property2):
        ...

    def __call__(
        self,
        func: httpx.request,
        method_data: MethodData,
    ):
        # some custom logic that handles response
        # result: httpx.Response = func(**method_data)

class AsyncMyRetryStrategy(AsyncBaseRetryStrategy):
    def __init__(self, custom_property1, custom_property2):
        ...

    async def __call__(
        self,
        func: httpx.request,
        method_data: MethodData,
    ):
        # some custom logic that handles response
        # result: httpx.Response = await func(**method_data)

API Methods

• search(query: str | None = None, name: str | None = None, country: str | None = None, format: Literal["json", "xml"] = "json", ) -> Response[SearchResponse] - Perform entity lookup based on the query, name and country

Parameters in looup:

• query - free text, can be anything
• name - free text, can be anything
• country - ISO country code like EN or NO for example

When performin search at least one parameter is needed

Types

We have following types:

class ParticipantId(TypedDict):
    """
    Secheme is identifier and value is combination of {issuing_agency_code}:{identififier}

    Scheme seems to always be `iso6523-actorid-upis`
    issuing_agency_code - numeric value assigned by ISO 6523 to specific national or
                          international registration authorities
    identififier - depends on issuing agency
    Some examples:
      - 0088: Global Location Number (GLN)
      - 0106: Dutch Chamber of Commerce (KVK)
      - 0192: Norwegian Organization Number (Organisasjonsnummer)
    """

    scheme: Literal["iso6523-actorid-upis"] | str
    value: str


class DocType(TypedDict):
    scheme: str
    value: str


class EntityName(TypedDict):
    """
    Name of the entity and language is ISO language code
    """

    name: str


class Entity(TypedDict):
    name: list[EntityName]
    countryCode: str
    regDate: str


class SearchMatch(TypedDict):
    participantID: ParticipantId
    docTypes: list[DocType]
    entities: list[Entity]


SearchResponse = TypedDict(
    "SearchResponse",
    {
        "version": str,
        "total-result-count": int,
        "used-result-count": int,
        "result-page-index": int,
        "result-page-count": int,
        "first-result-index": int,
        "last-result-index": int,
        "query-terms": str,
        "creation-dt": str,
        "matches": list[SearchMatch],
    },
)

Development

NOTE This project requires python3.12 or later

For development you need to clone project by running:

git clone {git source}

Once cloning is completed go to new folder and setup virtual environment

python3 -m venv ./venv
source ./venv/bin/activate
pip install -r requirements/dev.txt

This will install all packages that are needed for local development and testing

Once virtual environment is installed you can update scripts, or run tests

Testing

We are using pytest to run tests. All of the tests are located in ./src/tests. If file names are starting with test_ they will be automatically discovered. Also all of the functions that start with def test_ will be ran as tests.

To run tests you need to run following command:

pytest
# For verbose output
pytest -v

If you want to run tests in parallel to speed up execution (just remember rate limiting of 2req/s)

# Run with 10 workers, generally number of workers shouldn't exceed number of
# cpu cores
pytest -n 10

---

To generate coverage report you need to run following command:

pytest --cov --cov-report=html

This will run tests and generate coverage report. This will create folder htmlcov and put report there. You need to open htmlcov/index.html

In case you want just coverage report without tests you can run couple of commands:

# Generate command line report
coverage report

# Generate html report as in example above
coverage html

Note: You still need to run pytest --cov --cov-report=html before running commands above because with pytest command it will discover tests and generate .coverage file