refactored to use an ApiClient class

avt_fresh/api.py

@@ -0,0 +1,178 @@
+from avt_fresh import BASE_URL, REQUEST as _REQUEST
+from avt_fresh.client import (
+    FreshbooksClient,
+    get_freshbooks_client_from_email,
+    get_freshbooks_client_from_client_id,
+    get_freshbooks_client_from_org_name,
+    get_all_clients,
+    delete as delete_client,
+    create as create_client,
+    delete_contact,
+    add_contacts,
+)
+from avt_fresh.invoice import (
+    FreshbooksInvoice,
+    get_one as get_one_invoice,
+    get_all_draft_invoices,
+    get_all_invoices_for_client_id,
+    get_all_invoices_for_org_name,
+    get_draft_invoices_for_client_id,
+    create as create_invoice,
+    update as update_invoice,
+    delete as delete_invoice,
+    send as send_invoice,
+)
+from avt_fresh.payments import (
+    get_default_payment_options,
+    add_payment_option_to_invoice,
+)
+
+
+class ApiClient:
+    def __init__(
+        self, client_secret: str, client_id: str, redirect_uri: str, account_id: str
+    ):
+        self.client_secret = client_secret
+        self.client_id = client_id
+        self.redirect_uri = redirect_uri
+        self.account_id = account_id
+        self.url_lookup = self._make_url_lookup(account_id)
+
+    @staticmethod
+    def _make_url_lookup(account_id: str) -> dict[str, str]:
+        return {
+            "client": f"{BASE_URL}/accounting/account/{account_id}/users/clients",
+            "invoice": f"{BASE_URL}/accounting/account/{account_id}/invoices/invoices",
+            "payments": f"{BASE_URL}/payments/account/{account_id}",
+        }
+
+    def _REQUEST(
+        self, *, what: str, method_name: str, endpoint: str, stuff: dict | None = None
+    ) -> dict:
+        if method_name not in ("GET", "PUT", "POST"):
+            raise Exception
+        if what == "payments" and method_name == "PUT":
+            raise Exception
+
+        if what == "client" and method_name == "PUT":
+            url = f"{BASE_URL}/accounting/account"
+        else:
+            url = self.url_lookup[what]
+        return _REQUEST(
+            url=url, method_name=method_name, endpoint=endpoint, stuff=stuff
+        )
+
+    def _GET(self, *, what: str, endpoint: str, params=None):
+        return self._REQUEST(
+            what=what, method_name="GET", endpoint=endpoint, stuff=params
+        )
+
+    def _POST(self, *, what: str, endpoint: str, data: dict):
+        return self._REQUEST(
+            what=what, method_name="POST", endpoint=endpoint, stuff=data
+        )
+
+    def _PUT(self, *, what: str, thing_id: int, data: dict):
+        return self._REQUEST(
+            what=what, method_name="PUT", endpoint=f"/{thing_id}", stuff=data
+        )
+
+    def get_one_invoice(self, invoice_id: int) -> FreshbooksInvoice:
+        return get_one_invoice(get_func=self._GET, invoice_id=invoice_id)
+
+    def get_all_draft_invoices(self) -> list[FreshbooksInvoice]:
+        return get_all_draft_invoices(get_func=self._GET)
+
+    def get_all_invoices_for_org_name(self, org_name: str) -> list[FreshbooksInvoice]:
+        return get_all_invoices_for_org_name(get_func=self._GET, org_name=org_name)
+
+    def get_all_invoices_for_client_id(self, client_id: int) -> list[FreshbooksInvoice]:
+        return get_all_invoices_for_client_id(get_func=self._GET, client_id=client_id)
+
+    def get_draft_invoices_for_client_id(
+        self, client_id: int
+    ) -> list[FreshbooksInvoice]:
+        return get_draft_invoices_for_client_id(get_func=self._GET, client_id=client_id)
+
+    def create_invoice(
+        self,
+        *,
+        client_id: int,
+        notes: str,
+        lines: list[dict],
+        status: str | int,
+        contacts: list[dict] | None = None,
+        po_number=None,
+        create_date=None,
+    ) -> dict:
+        return create_invoice(
+            post_func=self._POST,
+            client_id=client_id,
+            notes=notes,
+            lines=lines,
+            status=status,
+            contacts=contacts,
+            po_number=po_number,
+            create_date=create_date,
+        )
+
+    def update_invoice(self, invoice_id: int, **kwargs) -> dict:
+        return update_invoice(put_func=self._PUT, invoice_id=invoice_id, **kwargs)
+
+    def delete_invoice(self, invoice_id: int) -> dict:
+        return delete_invoice(put_func=self._PUT, invoice_id=invoice_id)
+
+    def send_invoice(self, invoice_id: int) -> dict:
+        return send_invoice(put_func=self._PUT, invoice_id=invoice_id)
+
+    def get_freshbooks_client_from_email(self, email: str) -> FreshbooksClient:
+        return get_freshbooks_client_from_email(get_func=self._GET, email=email)
+
+    def get_freshbooks_client_from_client_id(self, client_id: int) -> FreshbooksClient:
+        return get_freshbooks_client_from_client_id(
+            get_func=self._GET, client_id=client_id
+        )
+
+    def get_freshbooks_client_from_org_name(self, org_name: str) -> FreshbooksClient:
+        return get_freshbooks_client_from_org_name(
+            get_func=self._GET, org_name=org_name
+        )
+
+    def get_all_clients(self) -> list[FreshbooksClient]:
+        return get_all_clients(get_func=self._GET)
+
+    def create_client(
+        self, first_name: str, last_name: str, email: str, organization: str
+    ) -> FreshbooksClient:
+        return create_client(
+            get_func=self._GET,
+            post_func=self._POST,
+            first_name=first_name,
+            last_name=last_name,
+            email=email,
+            organization=organization,
+        )
+
+    def delete_client(self, client_id: int) -> None:
+        delete_client(put_func=self._PUT, client_id=client_id)
+
+    def add_contacts(self, client_id: int, contacts: list[dict]) -> None:
+        add_contacts(
+            get_func=self._GET,
+            put_func=self._PUT,
+            client_id=client_id,
+            contacts=contacts,
+        )
+
+    def delete_contact(self, client_id: int, email: str) -> None:
+        delete_contact(
+            get_func=self._GET, put_func=self._PUT, client_id=client_id, email=email
+        )
+
+    def get_default_payment_options(self) -> dict:
+        return get_default_payment_options(get_func=self._GET)
+
+    def add_payment_option_to_invoice(self, invoice_id: int, gateway_name: str) -> dict:
+        return add_payment_option_to_invoice(
+            post_func=self._POST, invoice_id=invoice_id, gateway_name=gateway_name
+        )

avt_fresh/client.py

@@ -1,12 +1,9 @@
-from functools import partial
-from typing import NamedTuple
+import typing
 
-import requests
-
-from avt_fresh import BASE_URL, REQUEST as __REQUEST, ACCOUNT_ID, make_headers
+WHAT = "client"
 
 
-class FreshbooksContact(NamedTuple):
+class FreshbooksContact(typing.NamedTuple):
     contact_id: int
     first_name: str
     last_name: str
@@ -25,7 +22,7 @@ class FreshbooksContact(NamedTuple):
         }
 
 
-class FreshbooksClient(NamedTuple):
+class FreshbooksClient(typing.NamedTuple):
     client_id: int
     email: str
     organization: str
@@ -57,22 +54,6 @@ class FreshbooksClient(NamedTuple):
         )
 
 
-URL = f"{BASE_URL}/accounting/account/{ACCOUNT_ID}/users/clients"
-_REQUEST = partial(__REQUEST, url=URL)
-
-
-def _GET(endpoint, params=None):
-    return _REQUEST(method_name="GET", endpoint=endpoint, stuff=params)
-
-
-def _POST(endpoint, data: dict):
-    return _REQUEST(method_name="POST", endpoint=endpoint, stuff=data)
-
-
-def _PUT(endpoint, thing_id: int, data: dict):
-    return _REQUEST(method_name="PUT", endpoint=f"{endpoint}/{thing_id}", stuff=data)
-
-
 class NoResult(Exception):
     pass
 
@@ -84,59 +65,74 @@ class MoreThanOne(Exception):
 INCLUDE = "include[]=contacts"
 
 
-def get_freshbooks_client_from_email(email: str) -> FreshbooksClient:
+def get_freshbooks_client_from_email(
+    *, get_func: typing.Callable, email: str
+) -> FreshbooksClient:
     try:
-        return get_one(_GET(f"?search[email]={email}&{INCLUDE}"))
+        return _get_one(
+            get_func(what=WHAT, endpoint=f"?search[email]={email}&{INCLUDE}")
+        )
     except NoResult as e:
         raise NoResult(email) from e
 
 
-def get_freshbooks_client_from_org_name(org_name: str) -> FreshbooksClient:
-    return get_one(_GET(f"?search[organization_like]={org_name}&{INCLUDE}"))
-
-
-def get_freshbooks_client_from_client_id(client_id: int) -> FreshbooksClient:
-    return FreshbooksClient.from_api(**_GET(f"{client_id}?{INCLUDE}")["client"])
-
-
-def get_one(response: dict) -> FreshbooksClient:
-    clients = [FreshbooksClient.from_api(**client) for client in response["clients"]]
-    if len(clients) > 1:
-        print("warning, more than one result, returning the first")
-    elif not clients:
-        raise NoResult
-    return clients[0]
-
-
-def get_all_clients() -> list[FreshbooksClient]:
-    return _GET(f"?{INCLUDE}")
-
-
-def delete(client_id) -> None:
-    requests.put(
-        f"{BASE_URL}/accounting/account/{client_id}",
-        json={"client": {"vis_state": 1}},
-        headers=make_headers(),
+def get_freshbooks_client_from_org_name(
+    *, get_func: typing.Callable, org_name: str
+) -> FreshbooksClient:
+    return _get_one(
+        get_func(what=WHAT, endpoint=f"?search[organization_like]={org_name}&{INCLUDE}")
     )
 
 
+def get_freshbooks_client_from_client_id(
+    *, get_func: typing.Callable, client_id: int
+) -> FreshbooksClient:
+    return FreshbooksClient.from_api(
+        **get_func(what=WHAT, endpoint=f"{client_id}?{INCLUDE}")["client"]
+    )
+
+
+def get_all_clients(*, get_func: typing.Callable) -> list[FreshbooksClient]:
+    response = get_func(what=WHAT, endpoint="")
+    num_results = response["total"]
+    return [FreshbooksClient.from_api(**c) for c in get_func(what=WHAT, endpoint=f"?{INCLUDE}&per_page={num_results}")["clients"]]
+
+
+def delete(*, put_func: typing.Callable, client_id: int) -> None:
+    return put_func(what=WHAT, thing_id=client_id, data={"client": {"vis_state": 1}})
+
+
 def create(
-    first_name: str, last_name: str, email: str, organization: str
+    *,
+    get_func: typing.Callable,
+    post_func: typing.Callable,
+    first_name: str,
+    last_name: str,
+    email: str,
+    organization: str,
 ) -> FreshbooksClient:
     data = {
         "client": dict(
             fname=first_name, lname=last_name, email=email, organization=organization
         )
     }
-    client_id = _POST(endpoint="", data=data)["client"]["id"]
-    return get_freshbooks_client_from_client_id(client_id)
+    client_id = post_func(what=WHAT, endpoint="", data=data)["client"]["id"]
+    return get_freshbooks_client_from_client_id(get_func=get_func, client_id=client_id)
 
 
-def add_contacts(client_id, contacts: list[dict]) -> None:
+def add_contacts(
+    *,
+    get_func: typing.Callable,
+    put_func: typing.Callable,
+    client_id: int,
+    contacts: list[dict],
+) -> None:
     """contacts: [dict(email, fname, lname)]"""
     to_update = []
     new_contacts_email_dict = {c["email"]: c for c in contacts}
-    current_contacts = get_freshbooks_client_from_client_id(client_id).contacts
+    current_contacts = get_freshbooks_client_from_client_id(
+        get_func=get_func, client_id=client_id
+    ).contacts
 
     if current_contacts:
         for email, current_contact in current_contacts.items():
@@ -148,16 +144,21 @@ def add_contacts(client_id, contacts: list[dict]) -> None:
                 del new_contacts_email_dict[new_contact["email"]]
 
     to_update += list(new_contacts_email_dict.values())
-    update_contacts(client_id, to_update)
+    _update_contacts(put_func=put_func, client_id=client_id, contacts=to_update)
 
 
-def delete_contact(client_id, email) -> None:
-    client = get_freshbooks_client_from_client_id(client_id)
+def delete_contact(
+    *, get_func: typing.Callable, put_func: typing.Callable, client_id: int, email: str
+) -> None:
+    client = get_freshbooks_client_from_client_id(
+        get_func=get_func, client_id=client_id
+    )
     contact_to_delete = client.contacts.get(email)
     if contact_to_delete is not None:
-        return update_contacts(
-            client_id,
-            [
+        return _update_contacts(
+            put_func=put_func,
+            client_id=client_id,
+            contacts=[
                 client.dict
                 for client in client.contacts.values()
                 if client != contact_to_delete
@@ -165,13 +166,24 @@ def delete_contact(client_id, email) -> None:
         )
 
 
-def update_contacts(client_id, contacts: list[dict]) -> None:
-    _update_freshbooks_client(client_id, {"contacts": contacts})
-
-
-def _update_freshbooks_client(client_id, data: dict) -> None:
-    requests.put(
-        f"{BASE_URL}/accounting/account/{client_id}",
-        json={"client": data},
-        headers=make_headers(),
+def _update_contacts(
+    *, put_func: typing.Callable, client_id: int, contacts: list[dict]
+) -> None:
+    _update_freshbooks_client(
+        put_func=put_func, client_id=client_id, data={"contacts": contacts}
     )
+
+
+def _update_freshbooks_client(
+    *, put_func: typing.Callable, client_id: int, data: dict
+) -> None:
+    put_func(what=WHAT, thing_id=client_id, data={"client": data})
+
+
+def _get_one(response: dict) -> FreshbooksClient:
+    clients = [FreshbooksClient.from_api(**client) for client in response["clients"]]
+    if len(clients) > 1:
+        print("warning, more than one result, returning the first")
+    elif not clients:
+        raise NoResult
+    return clients[0]

avt_fresh/invoice.py

@@ -1,12 +1,9 @@
 import datetime as dt
 from decimal import Decimal
 from functools import partial, lru_cache
-from typing import NamedTuple
+import typing
 
-from avt_fresh import BASE_URL, ACCOUNT_ID, REQUEST as __REQUEST
-
-URL = f"{BASE_URL}/accounting/account/{ACCOUNT_ID}/invoices/invoices"
-_REQUEST = partial(__REQUEST, url=URL)
+WHAT = "invoice"
 
 
 class ArgumentError(Exception):
@@ -25,19 +22,7 @@ class InvalidField(Exception):
     pass
 
 
-def _GET(endpoint, params=None):
-    return _REQUEST(method_name="GET", endpoint=endpoint, stuff=params)
-
-
-def _POST(endpoint, data: dict):
-    return _REQUEST(method_name="POST", endpoint=endpoint, stuff=data)
-
-
-def _PUT(endpoint, thing_id: int, data: dict):
-    return _REQUEST(method_name="PUT", endpoint=f"{endpoint}/{thing_id}", stuff=data)
-
-
-class FreshbooksLine(NamedTuple):
+class FreshbooksLine(typing.NamedTuple):
     invoice_id: int
     client_id: int
     description: str
@@ -70,7 +55,7 @@ class FreshbooksLine(NamedTuple):
         }
 
 
-class FreshbooksInvoice(NamedTuple):
+class FreshbooksInvoice(typing.NamedTuple):
     lines: list[FreshbooksLine]
     notes: str
     client_id: int
@@ -123,25 +108,31 @@ class FreshbooksInvoice(NamedTuple):
         )
 
 
-def get_all_draft_invoices():
-    return get(status="draft")
+def get_all_draft_invoices(*, get_func: typing.Callable) -> list[FreshbooksInvoice]:
+    return _get(get_func=get_func, status="draft")
 
 
-def get_all_invoices_for_org_name(org_name: str) -> list[FreshbooksInvoice]:
-    return get(org_name=org_name)
+def get_all_invoices_for_org_name(
+    *, get_func: typing.Callable, org_name: str
+) -> list[FreshbooksInvoice]:
+    return _get(get_func=get_func, org_name=org_name)
 
 
 @lru_cache
-def get_all_invoices_for_client_id(client_id: int) -> list[FreshbooksInvoice]:
-    return get(client_id=client_id)
+def get_all_invoices_for_client_id(
+    *, get_func: typing.Callable, client_id: int
+) -> list[FreshbooksInvoice]:
+    return _get(get_func=get_func, client_id=client_id)
 
 
-def get_draft_invoices_for_client_id(client_id: int) -> list[FreshbooksInvoice]:
-    return get(client_id=client_id, status="draft")
+def get_draft_invoices_for_client_id(
+    *, get_func: typing.Callable, client_id: int
+) -> list[FreshbooksInvoice]:
+    return _get(get_func=get_func, client_id=client_id, status="draft")
 
 
-def get_one(invoice_id) -> FreshbooksInvoice:
-    invoices = get(invoice_id)
+def get_one(*, get_func: typing.Callable, invoice_id: int) -> FreshbooksInvoice:
+    invoices = _get(get_func=get_func, invoice_id=invoice_id)
     if invoices:
         if len(invoices) > 1:
             raise MoreThanOne
@@ -149,12 +140,14 @@ def get_one(invoice_id) -> FreshbooksInvoice:
     raise DoesntExist
 
 
-def get(
+def _get(
+    get_func: typing.Callable,
     invoice_id=None,
     client_id=None,
     org_name=None,
     status=None,
 ) -> list[FreshbooksInvoice]:
+    get_func = partial(get_func, what=WHAT)
 
     if client_id is not None and org_name is not None:
         raise ArgumentError("Please provide either client_id or org_name")
@@ -180,7 +173,7 @@ def get(
         full_url += f"{sep}search[v3_status]={status}"
         sep = "&"
 
-    response = _GET(full_url)
+    response = get_func(endpoint=full_url)
     try:
         num_results = response["total"]
     except KeyError:
@@ -195,7 +188,7 @@ def get(
             f"&include[]=lines&include[]=contacts&include[]=allowed_gateways"
         )
 
-    result = _GET(full_url)
+    result = get_func(endpoint=full_url)
     if "invoice" in result:
         return [FreshbooksInvoice.from_api(**result["invoice"])]
     invoices = result["invoices"]
@@ -221,6 +214,8 @@ STATUS_STRING_INT_LOOKUP = {
 
 
 def create(
+    *,
+    post_func: typing.Callable,
     client_id: int,
     notes: str,
     lines: list[dict],
@@ -256,16 +251,20 @@ def create(
     if po_number:
         data["invoice"]["po_number"] = po_number
 
-    return _POST("", data=data)
+    return post_func(what=WHAT, endpoint="", data=data)
 
 
-def update(invoice_id, **kwargs) -> dict:
-    return _PUT("", invoice_id, dict(invoice=kwargs))
+def update(*, put_func: typing.Callable, invoice_id, **kwargs) -> dict:
+    return put_func(what=WHAT, thing_id=invoice_id, data={"invoice": kwargs})
 
 
-def delete(invoice_id) -> dict:
-    return _PUT("", invoice_id, dict(invoice=dict(vis_state=1)))
+def delete(*, put_func: typing.Callable, invoice_id: int) -> dict:
+    return put_func(what=WHAT, thing_id=invoice_id, data={"invoice": {"vis_state": 1}})
 
 
-def send(invoice_id) -> dict:
-    return _PUT("", invoice_id, {"invoice": {"action_email": True}})
+def send(*, put_func: typing.Callable, invoice_id: int) -> dict:
+    return put_func(
+        what=WHAT,
+        thing_id=invoice_id,
+        data={"invoice": {"action_email": True}},
+    )

avt_fresh/payments.py

@@ -1,26 +1,18 @@
-from functools import partial
+import typing
 
-from . import BASE_URL, ACCOUNT_ID, REQUEST as __REQUEST
-
-URL = f"{BASE_URL}/payments/account/{ACCOUNT_ID}"
-_REQUEST = partial(__REQUEST, url=URL)
+WHAT = "payments"
 
 
-def _GET(endpoint, params=None):
-    return _REQUEST(method_name="GET", endpoint=endpoint, stuff=params)
+def get_default_payment_options(*, get_func: typing.Callable) -> dict:
+    return get_func(what=WHAT, endpoint="payment_options?entity_type=invoice")
 
 
-def _POST(endpoint, data: dict):
-    return _REQUEST(method_name="POST", endpoint=endpoint, stuff=data)
-
-
-def get_default_payment_options():
-    return _GET("payment_options?entity_type=invoice")
-
-
-def add_payment_option_to_invoice(invoice_id, gateway_name="stripe"):
-    return _POST(
-        f"invoice/{invoice_id}/payment_options",
+def add_payment_option_to_invoice(
+    post_func: typing.Callable, invoice_id: int, gateway_name: str = "stripe"
+) -> dict:
+    return post_func(
+        what=WHAT,
+        endpoint=f"invoice/{invoice_id}/payment_options",
         data={
             "gateway_name": gateway_name,
             "entity_id": invoice_id,