LD-1893 Create and expose "invoices" via our API

.pre-commit-config.yaml

@@ -0,0 +1,38 @@
+default_stages: [commit]
+
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.3.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+
+  - repo: https://github.com/asottile/pyupgrade
+    rev: v2.38.0
+    hooks:
+      - id: pyupgrade
+        args: [--py39-plus]
+
+  - repo: https://github.com/psf/black
+    rev: 22.8.0
+    hooks:
+      - id: black
+
+  - repo: https://github.com/PyCQA/isort
+    rev: 5.12.0
+    hooks:
+      - id: isort
+
+  - repo: https://github.com/PyCQA/flake8
+    rev: 5.0.4
+    hooks:
+      - id: flake8
+        args: ["--config=setup.cfg"]
+        additional_dependencies: [flake8-isort]
+
+# sets up .pre-commit-ci.yaml to ensure pre-commit dependencies stay up to date
+ci:
+  autoupdate_schedule: weekly
+  skip: []
+  submodules: false

.pylintrc

@@ -0,0 +1,15 @@
+[MASTER]
+load-plugins=pylint_django, pylint_celery
+django-settings-module=config.settings.local
+[FORMAT]
+max-line-length=120
+
+[MESSAGES CONTROL]
+disable=missing-docstring,invalid-name
+disable=logging-format-interpolation
+
+[DESIGN]
+max-parents=13
+
+[TYPECHECK]
+generated-members=REQUEST,acl_users,aq_parent,"[a-zA-Z]+_set{1,2}",save,delete

examples/usage_example.py

@@ -1,13 +1,13 @@
+import itertools
 import random
-from justpaid.api import JustPaidAPI
-from justpaid.schemas import UsageEvent, UsageEventRequest
-from justpaid.exceptions import JustPaidAPIException
-
-from datetime import datetime, timedelta, timezone
+import sys
 import time
 import uuid
-import sys
-import itertools
+from datetime import datetime, timedelta, timezone
+
+from justpaid.api import JustPaidAPI
+from justpaid.exceptions import JustPaidAPIException
+from justpaid.schemas import UsageEvent, UsageEventRequest
 
 # Initialize the API with your token
 api = JustPaidAPI(api_token="YOUR_API_TOKEN")
@@ -22,10 +22,13 @@ def get_all_billable_items():
     except JustPaidAPIException as e:
         print(f"An error occurred: {str(e)}")
 
+
 def get_billable_items_by_external_customer_id(external_customer_id: str):
     try:
         # Get billable items by external customer ID
-        billable_items_response = api.get_billable_items(external_customer_id=external_customer_id)
+        billable_items_response = api.get_billable_items(
+            external_customer_id=external_customer_id
+        )
         print("Billable Items:")
         print(billable_items_response.customers[0])
 
@@ -54,12 +57,12 @@ def ingest_usage_events_async(usage_events):
         start_time = time.time()
 
         # Show a loading icon while polling job status
-        spinner = itertools.cycle(['|', '/', '-', '\\'])
+        spinner = itertools.cycle(["|", "/", "-", "\\"])
 
         # Poll the job status until it is complete
         job_status = api.get_usage_data_batch_job_status(response.job_id)
         while job_status.status not in ["SUCCESS", "FAILED"]:
-            sys.stdout.write('\rProcessing ' + next(spinner))
+            sys.stdout.write("\rProcessing " + next(spinner))
             sys.stdout.flush()
             time.sleep(0.1)
             job_status = api.get_usage_data_batch_job_status(response.job_id)
@@ -69,7 +72,7 @@ def ingest_usage_events_async(usage_events):
         processing_time = end_time - start_time
 
         # Clear the loading icon
-        sys.stdout.write('\r')
+        sys.stdout.write("\r")
         sys.stdout.flush()
 
         # Check final job status
@@ -96,7 +99,6 @@ def ingest_usage_events_async(usage_events):
         print(f"An error occurred during ingestion: {str(e)}")
 
 
-
 if __name__ == "__main__":
     usage_events = []
     for i in range(3000):
@@ -108,14 +110,9 @@ def ingest_usage_events_async(usage_events):
             idempotency_key=str(uuid.uuid4()),
             timestamp=timestamp,
             event_value=1,
-            properties={
-                "test": "test"
-            }
+            properties={"test": "test"},
         )
         usage_events.append(usage_event)
 
     # Call the separate method to ingest usage events asynchronously
     ingest_usage_events_async(usage_events)
-
-
-

setup.cfg

@@ -0,0 +1,18 @@
+[flake8]
+# https://black.readthedocs.io/en/stable/guides/using_black_with_other_tools.html#flake8
+max-line-length = 120
+extend-ignore = E203, E501
+exclude =
+    .git,
+    __pycache__,
+    */migrations/*
+
+
+[isort]
+# https://black.readthedocs.io/en/stable/guides/using_black_with_other_tools.html#isort
+profile = black
+# By default, `isort` will ignore skip configuration when paths are explicitly provided.
+# In order for `pre-commit` to respect this configuration, `filter_files` needs to be set to true.
+# https://jugmac00.github.io/blog/isort-and-pre-commit-a-friendship-with-obstacles/
+filter_files = true
+skip_glob = */migrations/*

setup.py

@@ -1,11 +1,12 @@
-from setuptools import setup, find_packages
 import os
 
+from setuptools import find_packages, setup
+
 version = {}
 with open(os.path.join("src", "justpaid", "_version.py")) as f:
     exec(f.read(), version)
 
-with open("README.md", "r", encoding="utf-8") as fh:
+with open("README.md", encoding="utf-8") as fh:
     long_description = fh.read()
 
 setup(

src/justpaid/__init__.py

@@ -1,6 +1,24 @@
-from ._version import __version__
+from ._version import __version__  # noqa: F401
 from .api import JustPaidAPI
-from .schemas import BillableItem, UsageEvent, UsageEventRequest, UsageEventResponse, BillableItemsResponse
 from .exceptions import JustPaidAPIException
+from .schemas import (
+    BillableItem,
+    BillableItemsResponse,
+    Invoice,
+    InvoiceListResponse,
+    UsageEvent,
+    UsageEventRequest,
+    UsageEventResponse,
+)
 
-__all__ = ['JustPaidAPI', 'BillableItem', 'UsageEvent', 'UsageEventRequest', 'UsageEventResponse', 'BillableItemsResponse', 'JustPaidAPIException']
+__all__ = [
+    "JustPaidAPI",
+    "BillableItem",
+    "UsageEvent",
+    "UsageEventRequest",
+    "UsageEventResponse",
+    "BillableItemsResponse",
+    "JustPaidAPIException",
+    "InvoiceListResponse",
+    "Invoice",
+]

src/justpaid/_version.py

@@ -1 +1 @@
-__version__ = "0.1.4" 
+__version__ = "0.1.5"

src/justpaid/api.py

@@ -1,10 +1,19 @@
-import requests
 from typing import Optional
+
+import requests
+
+from .exceptions import JustPaidAPIException
 from .schemas import (
-    BillableItem, BillableItemCustomer, UsageDataBatchJobStatusResponse, UsageEventRequest, 
-    UsageEventResponse, BillableItemsResponse, UsageEventAsyncResponse
+    BillableItemCustomer,
+    BillableItemsResponse,
+    Invoice,
+    InvoiceListResponse,
+    UsageDataBatchJobStatusResponse,
+    UsageEventAsyncResponse,
+    UsageEventRequest,
+    UsageEventResponse,
 )
-from .exceptions import JustPaidAPIException
+
 
 class JustPaidAPI:
     BASE_URL = "https://api.justpaid.io/api/v1"
@@ -13,59 +22,110 @@ def __init__(self, api_token: str):
         self.api_token = api_token
         self.headers = {
             "Authorization": f"Bearer {self.api_token}",
-            "Content-Type": "application/json"
+            "Content-Type": "application/json",
         }
 
-    def get_billable_items(self, customer_id: Optional[str] = None, external_customer_id: Optional[str] = None) -> BillableItemsResponse:
+    def get_billable_items(
+        self,
+        customer_id: Optional[str] = None,
+        external_customer_id: Optional[str] = None,
+    ) -> BillableItemsResponse:
         url = f"{self.BASE_URL}/usage/items"
         params = {}
         if customer_id:
             params["customer_id"] = customer_id
         if external_customer_id:
             params["external_customer_id"] = external_customer_id
-        
+
         response = requests.get(url, headers=self.headers, params=params)
-        
+
         if response.status_code != 200:
-            raise JustPaidAPIException(f"API request failed with status {response.status_code}: {response.text}")
-
+            raise JustPaidAPIException(
+                f"API request failed with status {response.status_code}: {response.text}"
+            )
+
         try:
             customers = [BillableItemCustomer(**item) for item in response.json()]
         except KeyError as e:
             raise JustPaidAPIException(f"Missing expected field in response: {e}")
-        
+
         return BillableItemsResponse(customers=customers)
 
-    def ingest_usage_events(self, payload: UsageEventRequest ) -> UsageEventResponse:
+    def ingest_usage_events(self, payload: UsageEventRequest) -> UsageEventResponse:
         url = f"{self.BASE_URL}/usage/ingest"
 
         response = requests.post(url, headers=self.headers, json=payload.dict())
-        
+
         if response.status_code != 200:
-            raise JustPaidAPIException(f"API request failed with status {response.status_code}: {response.text}")
-
+            raise JustPaidAPIException(
+                f"API request failed with status {response.status_code}: {response.text}"
+            )
+
         return UsageEventResponse(**response.json())
-
 
-    def ingest_usage_events_async(self, payload: UsageEventRequest ) -> UsageEventResponse:
+    def ingest_usage_events_async(
+        self, payload: UsageEventRequest
+    ) -> UsageEventResponse:
         url = f"{self.BASE_URL}/usage/ingest-async"
 
         response = requests.post(url, headers=self.headers, json=payload.dict())
-        
+
         if response.status_code != 200:
-            raise JustPaidAPIException(f"API request failed with status {response.status_code}: {response.text}")
-
+            raise JustPaidAPIException(
+                f"API request failed with status {response.status_code}: {response.text}"
+            )
+
         return UsageEventAsyncResponse(**response.json())
-
 
-    def get_usage_data_batch_job_status(self, job_id: str) -> UsageDataBatchJobStatusResponse:
+    def get_usage_data_batch_job_status(
+        self, job_id: str
+    ) -> UsageDataBatchJobStatusResponse:
         url = f"{self.BASE_URL}/usage/job_status/{job_id}"
 
         response = requests.get(url, headers=self.headers)
-        
+
         if response.status_code != 200:
-            raise JustPaidAPIException(f"API request failed with status {response.status_code}: {response.text}")
-
+            raise JustPaidAPIException(
+                f"API request failed with status {response.status_code}: {response.text}"
+            )
+
         return UsageDataBatchJobStatusResponse(**response.json())
-
-
+
+    def get_invoices(
+        self,
+        invoice_status: Optional[str] = None,
+        customer: Optional[str] = None,
+        limit: Optional[int] = None,
+        offset: Optional[int] = None,
+    ) -> InvoiceListResponse:
+        url = f"{self.BASE_URL}/invoice/"
+        params = {}
+        if invoice_status:
+            params["invoice_status"] = invoice_status
+        if customer:
+            params["customer"] = customer
+        if limit:
+            params["limit"] = limit
+        if offset:
+            params["offset"] = offset
+
+        response = requests.get(url, headers=self.headers, params=params)
+
+        if response.status_code != 200:
+            raise JustPaidAPIException(
+                f"API request failed with status {response.status_code}: {response.text}"
+            )
+
+        return InvoiceListResponse(**response.json())
+
+    def get_invoice(self, invoice_uuid: str) -> Invoice:
+        url = f"{self.BASE_URL}/invoice/{invoice_uuid}"
+
+        response = requests.get(url, headers=self.headers)
+
+        if response.status_code != 200:
+            raise JustPaidAPIException(
+                f"API request failed with status {response.status_code}: {response.text}"
+            )
+
+        return Invoice(**response.json())

src/justpaid/exceptions.py

@@ -1,3 +1,4 @@
 class JustPaidAPIException(Exception):
     """Exception raised for errors in the JustPaid API."""
+
     pass