Merge pull request #6 from paywithextend/add-expense-management-endpoints

Author: byjoh

extend/client.py

@@ -124,5 +124,29 @@ async def put(self, url: str, data: Dict) -> Any:
             response.raise_for_status()
             return response.json()
 
+    async def patch(self, url: str, data: Dict) -> Any:
+        """Make a PATCH request to the Extend API.
+
+        Args:
+            url (str): The API endpoint path (e.g., "/virtualcards/{card_id}")
+            data (Dict): The JSON payload to send in the request body
+
+        Returns:
+            The JSON response from the API
+
+        Raises:
+            httpx.HTTPError: If the request fails
+            ValueError: If the response is not valid JSON
+        """
+        async with httpx.AsyncClient() as client:
+            response = await client.patch(
+                self.build_full_url(url),
+                headers=self.headers,
+                json=data,
+                timeout=httpx.Timeout(30)
+            )
+            response.raise_for_status()
+            return response.json()
+
     def build_full_url(self, url: Optional[str]):
         return f"https://{API_HOST}{url or ''}"

extend/extend.py

@@ -1,6 +1,7 @@
 from extend.resources.virtual_cards import VirtualCards
 from .client import APIClient
 from .resources.credit_cards import CreditCards
+from .resources.expense_data import ExpenseData
 from .resources.transactions import Transactions
 
 
@@ -29,3 +30,4 @@ def __init__(self, api_key: str, api_secret: str):
         self.credit_cards = CreditCards(self._api_client)
         self.virtual_cards = VirtualCards(self._api_client)
         self.transactions = Transactions(self._api_client)
+        self.expense_data = ExpenseData(self._api_client)

extend/resources/credit_cards.py

@@ -49,3 +49,21 @@ async def get_credit_cards(
         params = {k: v for k, v in params.items() if v is not None}
 
         return await self._request(method="get", params=params)
+
+    async def get_credit_card_detail(
+            self,
+            card_id: str
+    ) -> Dict:
+        """Get detailed information about a specific credit card.
+
+        Args:
+            card_id (str): The unique identifier of the credit card
+
+        Returns:
+            Dict: A dictionary containing the credit card details:
+
+        Raises:
+            httpx.HTTPError: If the request fails
+        """
+
+        return await self._request(method="get", path=f"/{card_id}")

extend/resources/expense_data.py

@@ -0,0 +1,259 @@
+from typing import Optional, Dict
+
+from extend.client import APIClient
+from .resource import Resource
+
+
+class ExpenseData(Resource):
+    @property
+    def _base_url(self) -> str:
+        return "/expensedata"
+
+    def __init__(self, api_client: APIClient):
+        super().__init__(api_client)
+
+    async def get_expense_categories(
+            self,
+            active: Optional[bool] = None,
+            required: Optional[bool] = None,
+            search: Optional[str] = None,
+            sort_field: Optional[str] = None,
+            sort_direction: Optional[str] = None,
+    ) -> Dict:
+        """Get a list of expense categories.
+
+        Args:
+            active (Optional[bool]): Show only active categories
+            required (Optional[bool]): Show only required categories
+            search (Optional[str]): Full-text search query for filtering categories
+            sort_field (Optional[str]): Field to sort by
+            sort_direction (Optional[str]): Direction of sorting ("asc" or "desc")
+
+        Returns:
+            Dict: A dictionary containing:
+                - expenseCategories: List of Expense Category objects
+
+        Raises:
+            httpx.HTTPError: If the request fails
+        """
+
+        params = {
+            "active": active,
+            "required": required,
+            "search": search,
+            "sortField": sort_field,
+            "sortDirection": sort_direction,
+        }
+
+        return await self._request(method="get", path=f"/categories", params=params)
+
+    async def get_expense_category(self, category_id: str) -> Dict:
+        """Get detailed information about a specific expense category.
+
+        Args:
+            category_id (str): The unique identifier of the expense category
+
+        Returns:
+            Dict: A dictionary containing the expense category details
+
+        Raises:
+            httpx.HTTPError: If the request fails or transaction not found
+        """
+        return await self._request(method="get", path=f"/categories/{category_id}")
+
+    async def get_expense_category_labels(
+            self,
+            category_id: str,
+            page: Optional[int] = None,
+            per_page: Optional[int] = None,
+            active: Optional[bool] = None,
+            search: Optional[str] = None,
+            sort_field: Optional[str] = None,
+            sort_direction: Optional[str] = None,
+    ) -> Dict:
+        """Get a paginated list of expense categories.
+
+        Args:
+            category_id (str): The unique identifier of the expense category
+            page (Optional[int]): The page number for pagination (1-based)
+            per_page (Optional[int]): Number of items per page
+            active (Optional[bool]): Show only active labels
+            search (Optional[str]): Full-text search query
+            sort_field (Optional[str]): Field to sort by (e.g., activeLabelNameAsc, name)
+            sort_direction (Optional[str]): Sort direction (asc, desc) for sortable fields
+
+        Returns:
+            Dict: A dictionary containing:
+                - expenseLabels: List of Expense Category Label objects
+                - pagination: Dictionary containing the following pagination stats:
+                    - page: Current page number
+                    - pageItemCount: Number of items per page
+                    - totalItems: Total number expense category labels across all pages
+                    - numberOfPages: Total number of pages
+
+        Raises:
+            httpx.HTTPError: If the request fails
+        """
+
+        params = {
+            "page": page,
+            "count": per_page,
+            "active": active,
+            "search": search,
+            "sortField": sort_field,
+            "sortDirection": sort_direction,
+        }
+
+        return await self._request(method="get", path=f"/categories/{category_id}/labels", params=params)
+
+    async def create_expense_category(
+            self,
+            name: str,
+            code: str,
+            required: bool,
+            active: Optional[bool] = None,
+            free_text_allowed: Optional[bool] = None,
+            integrator_enabled: Optional[bool] = None,
+            integrator_field_number: Optional[int] = None,
+    ) -> Dict:
+        """Create an expense category.
+
+        Args:
+            name (str): User-facing name for this expense category
+            code (str): Code for the expense category
+            required (bool): Whether this field is required for all users
+            active (Optional[bool]): Whether this category is active and available for input
+            free_text_allowed (Optional[bool]): Whether free text input is allowed
+            integrator_enabled (Optional[bool]): Whether this category is integrator enabled
+            integrator_field_number (Optional[int]): Field number used by the integrator
+
+        Returns:
+            Dict: A dictionary containing the newly created expense category
+
+        Raises:
+            httpx.HTTPError: If the request fails or the transaction is not found
+        """
+
+        payload = {
+            "name": name,
+            "code": code,
+            "required": required,
+            "active": active,
+            "freeTextAllowed": free_text_allowed,
+            "integratorEnabled": integrator_enabled,
+            "integratorFieldNumber": integrator_field_number,
+        }
+
+        return await self._request(
+            method="post",
+            params=payload
+        )
+
+    async def create_expense_category_label(
+            self,
+            category_id: str,
+            name: str,
+            code: str,
+            active: bool = True
+    ) -> Dict:
+        """Create an expense category.
+
+        Args:
+            category_id (str): The unique identifier of the expense category
+            name (str): User-facing name for this expense category label
+            code (str): Code for the expense category label
+            active (bool): Whether the label is active and available for input
+
+        Returns:
+            Dict: A dictionary containing the newly created expense category label
+
+        Raises:
+            httpx.HTTPError: If the request fails or the transaction is not found
+        """
+        payload = {
+            "name": name,
+            "code": code,
+            "active": active,
+        }
+
+        return await self._request(
+            method="post",
+            path=f"/{category_id}",
+            params=payload
+        )
+
+    async def update_expense_category(
+            self,
+            category_id: str,
+            name: Optional[str] = None,
+            active: Optional[bool] = None,
+            required: Optional[bool] = None,
+            free_text_allowed: Optional[bool] = None,
+            integrator_enabled: Optional[bool] = None,
+            integrator_field_number: Optional[int] = None,
+    ) -> Dict:
+        """Update the an expense category.
+
+        Args:
+            category_id (str): The unique identifier of the expense category
+            name (Optional[str]): User-facing name for this expense category
+            active (Optional[bool]): Whether the category is active
+            required (Optional[bool]): Whether this field is required for all users
+            free_text_allowed (Optional[bool]): Whether free text input is allowed
+            integrator_enabled (Optional[bool]): Whether this category is integrator enabled
+            integrator_field_number (Optional[int]): Field number used by the integrator
+
+        Returns:
+            Dict: A dictionary containing the updated expense category details
+
+        Raises:
+            httpx.HTTPError: If the request fails or the transaction is not found
+        """
+
+        payload = {
+            "name": name,
+            "active": active,
+            "required": required,
+            "freeTextAllowed": free_text_allowed,
+            "integratorEnabled": integrator_enabled,
+            "integratorFieldNumber": integrator_field_number,
+        }
+
+        return await self._request(
+            method="patch",
+            path=f"/{category_id}",
+            params=payload
+        )
+
+    async def update_expense_category_label(
+            self,
+            category_id: str,
+            label_id: str,
+            name: Optional[str] = None,
+            active: Optional[bool] = None,
+    ) -> Dict:
+        """Update an expense category label.
+
+        Args:
+            category_id (str): The unique identifier of the expense category
+            label_id (str): The unique identifier of the expense category label to update
+            name (Optional[str]): User-facing name for the expense label
+            active (Optional[bool]): Whether the label is active and available for input
+
+        Returns:
+            Dict: A dictionary containing the updated expense category details
+
+        Raises:
+            httpx.HTTPError: If the request fails or the transaction is not found
+        """
+
+        payload = {
+            "name": name,
+            "active": active,
+        }
+
+        return await self._request(
+            method="patch",
+            path=f"/{category_id}/labels/{label_id}",
+            params=payload
+        )

extend/resources/resource.py

@@ -22,13 +22,17 @@ async def _request(
             path: str = None,
             params: Optional[Dict] = None
     ) -> Any:
+        if params is not None:
+            params = {k: v for k, v in params.items() if v is not None}
         match method:
             case "get":
                 return await self._api_client.get(self.build_full_path(path), params)
             case "post":
                 return await self._api_client.post(self.build_full_path(path), params)
             case "put":
                 return await self._api_client.put(self.build_full_path(path), params)
+            case "patch":
+                return await self._api_client.patch(self.build_full_path(path), params)
             case _:
                 raise ValueError(f"Unsupported HTTP method: {method}")
 

extend/resources/transactions.py

@@ -75,3 +75,39 @@ async def get_transaction(self, transaction_id: str) -> Dict:
             httpx.HTTPError: If the request fails or transaction not found
         """
         return await self._request(method="get", path=f"/{transaction_id}")
+
+    async def update_transaction_expense_data(self, transaction_id: str, data: Dict) -> Dict:
+        """Update the expense data for a specific transaction.
+
+        Args:
+            transaction_id (str): The unique identifier of the transaction
+            data (Dict): A dictionary representing the expense data to update, should match
+                         the schema:
+                         {
+                             "supplier": {
+                                 "name": "Some Supplier",
+                                 "id": "supplier-id"
+                             },
+                             "expenseCategories": [
+                                 {
+                                     "categoryCode": "COMPCODE",
+                                     "labelCode": "ABC123"
+                                 }
+                             ],
+                             "customer": {
+                                 "name": "Some Customer",
+                                 "id": "customer-id"
+                             }
+                         }
+
+        Returns:
+            Dict: A dictionary containing the updated transaction details
+
+        Raises:
+            httpx.HTTPError: If the request fails or the transaction is not found
+        """
+        return await self._request(
+            method="patch",
+            path=f"/{transaction_id}/expensedata",
+            params=data
+        )

extend/resources/virtual_cards.py

@@ -137,8 +137,8 @@ async def create_virtual_card(
     async def update_virtual_card(
             self,
             card_id: str,
+            display_name: str,
             balance_cents: int,
-            display_name: Optional[str] = None,
             notes: Optional[str] = None,
             valid_from: Optional[str] = None,
             valid_to: Optional[str] = None