Elorus API client for python

Author: Oluwafemi Adenuga

.devcontainer/devcontainer.json

@@ -0,0 +1,44 @@
+{
+    "name": "elorus-python-sdk",
+    "dockerComposeFile": [
+      "../docker-compose.yml"
+    ],
+    "service": "elorus-python-sdk",
+    "workspaceFolder": "/usr/src/app",
+    "features": {
+      "ghcr.io/devcontainers/features/git:1": {
+        "ppa": true,
+        "version": "os-provided"
+      }
+    },
+    "customizations": {
+      "vscode": {
+        "extensions": [
+          "editorconfig.editorconfig",
+          "ms-python.black-formatter",
+          "ms-python.isort",
+          "ms-python.python",
+          "ms-python.vscode-pylance"
+        ],
+        "settings": {
+          "editor.formatOnPaste": true,
+          "editor.formatOnSave": true,
+          "editor.formatOnSaveMode": "modifications",
+          "isort.args": [
+            "--profile",
+            "black"
+          ],
+          "python.formatting.provider": "black",
+          "[python]": {
+            "editor.defaultFormatter": "ms-python.black-formatter",
+            "editor.formatOnSave": true,
+            "editor.formatOnSaveMode": "modifications",
+            "editor.formatOnPaste": true,
+            "editor.codeActionsOnSave": {
+              "source.organizeImports": "always"
+            }
+          }
+        }
+      }
+    }
+  }

.github/ci.yml

@@ -0,0 +1,21 @@
+name: CI
+
+on:
+  pull_request:
+  push:
+    branches:
+      - main
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/[email protected]
+      - run: pipx install poetry==1.8.2
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: poetry
+      - run: poetry install
+      - run: poetry run black --check .
+      - run: poetry run mypy --check .

.github/publish.yml

@@ -0,0 +1,22 @@
+name: Publish to PyPI
+
+on:
+  release:
+    types:
+      - published
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/[email protected]
+      - run: pipx install poetry==1.8.2
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: "poetry"
+      - run: poetry install
+      - name: Build and publish
+        run: |
+          poetry config pypi-token.pypi ${{ secrets.PYPI_TOKEN }}
+          poetry publish --build

.gitignore

@@ -158,3 +158,4 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
+run.py

Dockerfile

@@ -0,0 +1,8 @@
+FROM ghcr.io/withlogicco/poetry:1.8.2-python-3.12
+
+WORKDIR /usr/src/app
+COPY pyproject.toml poetry.lock README.md ./
+RUN poetry check && poetry lock --check
+RUN poetry install
+
+COPY ./ ./

README.md

@@ -1 +1,7 @@
-# elorus-python-sdk
+# Elorus Python SDK
+
+`Elorus` is a Python SDK for interacting with the API of [Elorus](https://developer.elorus.com/).
+
+## Requirements
+
+Python 3.12 or later

docker-compose.yml

@@ -0,0 +1,14 @@
+version: "3.8"
+
+services:
+  elorus-python-sdk:
+    build: .
+    environment:
+      ELORUS_API_KEY: ${ELORUS_API_KEY}
+      ELORUS_ORGANIZATION_ID: ${ELORUS_ORGANIZATION_ID}
+      ELORUS_BASE_URL: ${ELORUS_BASE_URL}
+    volumes:
+      - .:/usr/src/app
+    network_mode: host
+    image: elorus-python-sdk:latest
+    command: sleep infinity

elorus/__init__.py

@@ -0,0 +1,15 @@
+from httpx import Auth
+
+
+class ElorusAuthentication(Auth):
+    def __init__(self, token: str, elorus_organization_id: str, is_demo: bool = False):
+        self.token = token
+        self.elorus_organization_id = elorus_organization_id
+        self.is_demo = is_demo
+
+    def auth_flow(self, request):
+        request.headers["Authorization"] = f"Token {self.token}"
+        request.headers["X-Elorus-Organization"] = self.elorus_organization_id
+        if self.is_demo:
+            request.headers["X-Elorus-Demo"] = "true"
+        yield request

elorus/auth.py

@@ -0,0 +1,191 @@
+from dataclasses import asdict
+from typing import Optional
+
+import httpx
+
+from elorus.auth import ElorusAuthentication
+from elorus.exceptions import (
+    AuthenticationError,
+    AuthorizationError,
+    BadRequestError,
+    Error,
+    ThrottlingError,
+)
+from elorus.models import Contact, EmailBody, Invoice
+
+
+class Client:
+    def __init__(
+        self,
+        api_key: str,
+        elorus_organization_id: str,
+        is_demo: bool = False,
+        base_url: Optional[str] = "https://api.elorus.com",
+        api_version: Optional[str] = "v1.1",
+    ):
+        self.base_url = base_url
+        self.api_key = api_key
+        self.elorus_organization_id = elorus_organization_id
+        self.is_demo = is_demo
+        self.api_version = api_version
+
+        self.contacts = Contacts(self)
+        self.invoices = Invoices(self)
+
+    def _get_auth(self):
+        return ElorusAuthentication(
+            token=self.api_key,
+            elorus_organization_id=self.elorus_organization_id,
+            is_demo=self.is_demo,
+        )
+
+    def handle_file_download(self, response):
+        content_disposition = response.headers.get("Content-Disposition", "")
+        if "filename=" not in content_disposition:
+            raise ValueError("No filename found in Content-Disposition header")
+        filename_kv = content_disposition.split("filename=")
+        if len(filename_kv) < 2:
+            raise ValueError("Invalid Content-Disposition header format")
+        filename = filename_kv[1]
+        return filename, response.content
+
+    def _handle_response(self, response):
+        if response.headers.get("Content-Type", "").lower() == "application/pdf":
+            return self.handle_file_download(response)
+
+        if response.status_code == 204:
+            return response.text
+
+        message = response.json()
+
+        if response.status_code == 401:
+            raise AuthenticationError(
+                message=message,
+                response=response,
+            )
+
+        if response.status_code == 403:
+            raise AuthorizationError(
+                message=message,
+                response=response,
+            )
+
+        if response.status_code == 429:
+            raise ThrottlingError(
+                message=message,
+                response=response,
+            )
+
+        if response.status_code == 400:
+            raise BadRequestError(
+                message=message,
+                response=response,
+            )
+
+        if response.status_code >= 500:
+            raise Error(
+                message=message,
+                response=response,
+            )
+
+        try:
+            response.raise_for_status()
+            return message
+        except:
+            error = f"{response.text}"
+            content_type = response.headers.get("Content-Type", "").lower()
+            if content_type == "application/json":
+                resp = response.json()
+                error_message_keys = ("message", "msg", "detail")
+                error_message = next(
+                    (resp[key] for key in error_message_keys if key in resp), None
+                )
+                if error_message:
+                    error = f"Message: {error_message} , Error details: {resp.get('errors')}, {resp.get('data')}"
+                    raise Error(error, response)
+
+            raise Error(error, response)
+
+    def _handle_request(
+        self, method: str, path: str, payload: Optional[dict] = None, **kwargs
+    ):
+        auth = self._get_auth()
+        url = f"{self.base_url}/{self.api_version}/{path}"
+        with httpx.Client(auth=auth) as client:
+            response = client.request(method, url, json=payload, **kwargs)
+            return self._handle_response(response)
+
+
+class SubClient:
+    client = Client
+
+    def __init__(self, client: Client):
+        self.client = client
+
+
+class Contacts(SubClient):
+
+    def list(self):
+        return self.client._handle_request("GET", "contacts/")
+
+    def create(self, contact: Contact):
+        payload = asdict(contact)
+        return self.client._handle_request("POST", "contacts/", payload=payload)
+
+    def get(self, contact_id: str):
+        return self.client._handle_request("GET", f"contacts/{contact_id}/")
+
+    def update(self, contact_id: str, contact: Contact):
+        payload = asdict(contact)
+        return self.client._handle_request(
+            "PUT", f"contacts/{contact_id}/", payload=payload
+        )
+
+    def delete(self, contact_id: str):
+        return self.client._handle_request("DELETE", f"contacts/{contact_id}/")
+
+
+class Invoices(SubClient):
+
+    def list(self):
+        return self.client._handle_request("GET", "invoices/")
+
+    def create(self, invoice: Invoice):
+        payload = invoice.clean_dict()
+        return self.client._handle_request("POST", "invoices/", payload=payload)
+
+    def get(self, invoice_id: str):
+        return self.client._handle_request("GET", f"invoices/{invoice_id}/")
+
+    def update(self, invoice_id: str, invoice: Invoice):
+        payload = invoice.clean_dict()
+        return self.client._handle_request(
+            "PUT", f"invoices/{invoice_id}/", payload=payload
+        )
+
+    def partial_update(self, invoice_id: str, invoice: Invoice):
+        payload = invoice.clean_dict()
+        return self.client._handle_request(
+            "PATCH", f"invoices/{invoice_id}/", payload=payload
+        )
+
+    def delete(self, invoice_id: str):
+        return self.client._handle_request("DELETE", f"invoices/{invoice_id}/")
+
+    def post_email(self, invoice_id: str, email_body: EmailBody):
+        payload = email_body.clean_dict()
+        return self.client._handle_request(
+            "POST", f"invoices/{invoice_id}/email/", payload=payload
+        )
+
+    def get_email(self, invoice_id: str):
+        return self.client._handle_request("GET", f"invoices/{invoice_id}/email/")
+
+    def get_pdf(self, invoice_id: str):
+        return self.client._handle_request("GET", f"invoices/{invoice_id}/pdf/")
+
+    def mark_void(self, invoice_id: str, is_void: bool = True):
+        payload = {"is_void": is_void}
+        return self.client._handle_request(
+            "POST", f"invoices/{invoice_id}/void/", payload=payload
+        )

elorus/client.py

@@ -0,0 +1,37 @@
+class Error(Exception):
+    """Base class for other exceptions"""
+
+    def __init__(self, message, response=None):
+        self.message = message
+        self.response = response
+
+    def __str__(self) -> str:
+        return f"Error code: {self.response.status_code}, Error message: {self.message}"
+
+
+class AuthenticationError(Error):
+    """Raised when there is an authentication error"""
+
+    def __init__(self, message, response=None):
+        super().__init__(message, response)
+
+
+class AuthorizationError(Error):
+    """Raised when there is an authorization error"""
+
+    def __init__(self, message, response=None):
+        super().__init__(message, response)
+
+
+class BadRequestError(Error):
+    """Raised when the request is bad"""
+
+    def __init__(self, message, response=None):
+        super().__init__(message, response)
+
+
+class ThrottlingError(Error):
+    """Raised when the rate limit is exceeded"""
+
+    def __init__(self, message, response=None):
+        super().__init__(message, response)