Add Petition Decisions API client

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 
+## [0.1.3] - 2024-05-06
+### Added
+- Support for the Final Petition Decisions API
+
 ## [0.1.2]
 
 ### Added

README.md

@@ -6,8 +6,7 @@
 
 A Python client library for interacting with the United Stated Patent and Trademark Office (USPTO) [Open Data Portal](https://data.uspto.gov/home) APIs.
 
-This package provides clients for interacting with both the USPTO Bulk Data API and the USPTO Patent Data API.
-The client for the Final Petition Decisions API is currently being developed. 
+This package provides clients for interacting with the USPTO Bulk Data API, the USPTO Patent Data API, and the Final Petition Decisions API.
 
 > [!IMPORTANT]
 > The USPTO is in the process of moving their API. This package is only concerned with the new API. The [old API](https://developer.uspto.gov/) will be retired at the end of 2025.
@@ -72,7 +71,7 @@ print(f"Found {inventor_search.count} applications with 'Smith' as inventor")
 
 ## Features
 
-- Access to both USPTO Bulk Data API and Patent Data API
+- Access to the USPTO Bulk Data API, Patent Data API, and Final Petition Decisions API
 - Search for patent applications using various filters
 - Download files and documents from the APIs
 

docs/source/api/clients.rst

@@ -10,3 +10,8 @@ Clients
    :members:
    :undoc-members:
    :show-inheritance:
+
+.. automodule:: pyUSPTO.clients.petition_decisions
+   :members:
+   :undoc-members:
+   :show-inheritance:

docs/source/api/models.rst

@@ -10,3 +10,8 @@ Models
    :members:
    :undoc-members:
    :show-inheritance:
+
+.. automodule:: pyUSPTO.models.petition_decisions
+   :members:
+   :undoc-members:
+   :show-inheritance:

docs/source/examples/index.rst

@@ -6,3 +6,4 @@ Examples
 
    bulk_data
    patent_data
+   petition_decisions

docs/source/examples/petition_decisions.rst

@@ -0,0 +1,6 @@
+Petition Decisions Examples
+==========================
+
+.. literalinclude:: ../../examples/petition_decisions_example.py
+   :language: python
+   :linenos:

examples/petition_decisions_example.py

@@ -0,0 +1,18 @@
+"""Example usage of the PetitionDecisionsClient."""
+
+import json
+import os
+
+from pyUSPTO.clients.petition_decisions import PetitionDecisionsClient
+from pyUSPTO.config import USPTOConfig
+
+# Initialize client using API key from environment or placeholder
+api_key = os.environ.get("USPTO_API_KEY", "YOUR_API_KEY_HERE")
+client = PetitionDecisionsClient(api_key=api_key)
+
+# Search for decisions containing 'highway'
+try:
+    response = client.search_decisions(query="highway", limit=5)
+    print(json.dumps(response.to_dict(), indent=2))
+except Exception as exc:  # pragma: no cover - example usage
+    print(f"Error fetching decisions: {exc}")

src/pyUSPTO/__init__.py

@@ -14,6 +14,7 @@
 
 from pyUSPTO.clients.bulk_data import BulkDataClient
 from pyUSPTO.clients.patent_data import PatentDataClient
+from pyUSPTO.clients.petition_decisions import PetitionDecisionsClient
 from pyUSPTO.config import USPTOConfig
 from pyUSPTO.exceptions import (
     USPTOApiAuthError,
@@ -30,6 +31,10 @@
     ProductFileBag,
 )
 from pyUSPTO.models.patent_data import PatentDataResponse, PatentFileWrapper
+from pyUSPTO.models.petition_decisions import (
+    PetitionDecision,
+    PetitionDecisionsResponse,
+)
 
 __all__ = [
     # Base classes
@@ -48,4 +53,8 @@
     "PatentDataClient",
     "PatentDataResponse",
     "PatentFileWrapper",
+    # Petition Decisions API
+    "PetitionDecisionsClient",
+    "PetitionDecisionsResponse",
+    "PetitionDecision",
 ]

src/pyUSPTO/clients/__init__.py

@@ -6,8 +6,10 @@
 
 from pyUSPTO.clients.bulk_data import BulkDataClient
 from pyUSPTO.clients.patent_data import PatentDataClient
+from pyUSPTO.clients.petition_decisions import PetitionDecisionsClient
 
 __all__ = [
     "BulkDataClient",
     "PatentDataClient",
+    "PetitionDecisionsClient",
 ]

src/pyUSPTO/clients/petition_decisions.py

@@ -0,0 +1,74 @@
+"""Client for the USPTO Petition Decisions API."""
+
+from typing import Any, Dict, Iterator, Optional
+
+from pyUSPTO.clients.base import BaseUSPTOClient
+from pyUSPTO.config import USPTOConfig
+from pyUSPTO.models.petition_decisions import (
+    PetitionDecision,
+    PetitionDecisionsResponse,
+)
+
+
+class PetitionDecisionsClient(BaseUSPTOClient[PetitionDecisionsResponse]):
+    """Client for interacting with the Petition Decisions API."""
+
+    ENDPOINTS = {
+        "search_decisions": "api/v1/petition/decisions/search",
+    }
+
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        base_url: Optional[str] = None,
+        config: Optional[USPTOConfig] = None,
+    ) -> None:
+        self.config = config or USPTOConfig(api_key=api_key)
+        api_key_to_use = api_key or self.config.api_key
+        effective_base_url = (
+            base_url
+            or self.config.petition_decisions_base_url
+            or "https://api.uspto.gov"
+        )
+        super().__init__(api_key=api_key_to_use, base_url=effective_base_url)
+
+    def search_decisions(
+        self,
+        query: Optional[str] = None,
+        offset: Optional[int] = None,
+        limit: Optional[int] = None,
+        facets: Optional[bool] = None,
+        additional_query_params: Optional[Dict[str, Any]] = None,
+    ) -> PetitionDecisionsResponse:
+        """Search for petition decisions."""
+
+        params: Dict[str, Any] = {}
+        if query:
+            params["q"] = query
+        if offset is not None:
+            params["offset"] = offset
+        if limit is not None:
+            params["limit"] = limit
+        if facets is not None:
+            params["facets"] = str(facets).lower()
+        if additional_query_params:
+            params.update(additional_query_params)
+
+        result = self._make_request(
+            method="GET",
+            endpoint=self.ENDPOINTS["search_decisions"],
+            params=params or None,
+            response_class=PetitionDecisionsResponse,
+        )
+        assert isinstance(result, PetitionDecisionsResponse)
+        return result
+
+    def paginate_decisions(self, **kwargs: Any) -> Iterator[PetitionDecision]:
+        """Paginate through all decisions matching the search criteria."""
+
+        return self.paginate_results(
+            method_name="search_decisions",
+            response_container_attr="petition_decision_data_bag",
+            **kwargs,
+        )
+

src/pyUSPTO/config.py

@@ -16,6 +16,7 @@ def __init__(
         api_key: Optional[str] = None,
         bulk_data_base_url: str = "https://api.uspto.gov",
         patent_data_base_url: str = "https://api.uspto.gov",
+        petition_decisions_base_url: str = "https://api.uspto.gov",
     ):
         """
         Initialize the USPTOConfig.
@@ -24,13 +25,15 @@ def __init__(
             api_key: API key for authentication, defaults to USPTO_API_KEY environment variable
             bulk_data_base_url: Base URL for the Bulk Data API
             patent_data_base_url: Base URL for the Patent Data API
+            petition_decisions_base_url: Base URL for the Petition Decisions API
         """
         # Use environment variable only if api_key is None, not if it's an empty string
         self.api_key = (
             api_key if api_key is not None else os.environ.get("USPTO_API_KEY")
         )
         self.bulk_data_base_url = bulk_data_base_url
         self.patent_data_base_url = patent_data_base_url
+        self.petition_decisions_base_url = petition_decisions_base_url
 
     @classmethod
     def from_env(cls) -> "USPTOConfig":
@@ -48,4 +51,7 @@ def from_env(cls) -> "USPTOConfig":
             patent_data_base_url=os.environ.get(
                 "USPTO_PATENT_DATA_BASE_URL", "https://api.uspto.gov"
             ),
+            petition_decisions_base_url=os.environ.get(
+                "USPTO_PETITION_DECISIONS_BASE_URL", "https://api.uspto.gov"
+            ),
         )

src/pyUSPTO/models/petition_decisions.py

@@ -0,0 +1,117 @@
+"""Data models for the Petition Decisions API."""
+
+from dataclasses import asdict, dataclass, field
+from datetime import date, datetime
+from typing import Any, Dict, List, Optional
+
+from pyUSPTO.models.patent_data import (
+    parse_to_date,
+    parse_to_datetime_utc,
+    serialize_date,
+    serialize_datetime_as_iso,
+    to_camel_case,
+)
+
+
+@dataclass(frozen=True)
+class PetitionDecision:
+    """Represents a single petition decision record."""
+
+    action_taken_by_court_name: Optional[str] = None
+    application_number_text: Optional[str] = None
+    business_entity_status_category: Optional[str] = None
+    court_action_indicator: Optional[bool] = None
+    customer_number: Optional[int] = None
+    decision_date: Optional[date] = None
+    decision_petition_type_code: Optional[int] = None
+    decision_type_code: Optional[str] = None
+    decision_type_code_description_text: Optional[str] = None
+    final_deciding_office_name: Optional[str] = None
+    first_applicant_name: Optional[str] = None
+    first_inventor_to_file_indicator: Optional[bool] = None
+    group_art_unit_number: Optional[str] = None
+    invention_title: Optional[str] = None
+    inventor_bag: List[str] = field(default_factory=list)
+    last_ingestion_date_time: Optional[datetime] = None
+    petition_decision_record_identifier: Optional[str] = None
+    petition_issue_considered_text_bag: List[str] = field(default_factory=list)
+    petition_mail_date: Optional[date] = None
+    rule_bag: List[str] = field(default_factory=list)
+    technology_center: Optional[str] = None
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "PetitionDecision":
+        """Create a :class:`PetitionDecision` from API data."""
+        return cls(
+            action_taken_by_court_name=data.get("actionTakenByCourtName"),
+            application_number_text=data.get("applicationNumberText"),
+            business_entity_status_category=data.get("businessEntityStatusCategory"),
+            court_action_indicator=data.get("courtActionIndicator"),
+            customer_number=data.get("customerNumber"),
+            decision_date=parse_to_date(data.get("decisionDate")),
+            decision_petition_type_code=data.get("decisionPetitionTypeCode"),
+            decision_type_code=data.get("decisionTypeCode"),
+            decision_type_code_description_text=data.get("decisionTypeCodeDescriptionText"),
+            final_deciding_office_name=data.get("finalDecidingOfficeName"),
+            first_applicant_name=data.get("firstApplicantName"),
+            first_inventor_to_file_indicator=data.get("firstInventorToFileIndicator"),
+            group_art_unit_number=data.get("groupArtUnitNumber"),
+            invention_title=data.get("inventionTitle"),
+            inventor_bag=data.get("inventorBag", []),
+            last_ingestion_date_time=parse_to_datetime_utc(data.get("lastIngestionDateTime")),
+            petition_decision_record_identifier=data.get("petitionDecisionRecordIdentifier"),
+            petition_issue_considered_text_bag=data.get("petitionIssueConsideredTextBag", []),
+            petition_mail_date=parse_to_date(data.get("petitionMailDate")),
+            rule_bag=data.get("ruleBag", []),
+            technology_center=data.get("technologyCenter"),
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert the decision to a dictionary with camelCase keys."""
+
+        d = asdict(self)
+        d["decision_date"] = serialize_date(self.decision_date)
+        d["petition_mail_date"] = serialize_date(self.petition_mail_date)
+        d["last_ingestion_date_time"] = serialize_datetime_as_iso(
+            self.last_ingestion_date_time
+        )
+        return {
+            to_camel_case(k): v
+            for k, v in d.items()
+            if v is not None and (not isinstance(v, list) or v)
+        }
+
+
+@dataclass(frozen=True)
+class PetitionDecisionsResponse:
+    """Top level response from the Petition Decisions API."""
+
+    count: int
+    request_identifier: Optional[str] = None
+    petition_decision_data_bag: List[PetitionDecision] = field(default_factory=list)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "PetitionDecisionsResponse":
+        """Create a :class:`PetitionDecisionsResponse` from API data."""
+        return cls(
+            count=data.get("count", 0),
+            request_identifier=data.get("requestIdentifier"),
+            petition_decision_data_bag=[
+                PetitionDecision.from_dict(d)
+                for d in data.get("petitionDecisionDataBag", [])
+                if isinstance(d, dict)
+            ],
+        )
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert this response object back to a dictionary."""
+        d = asdict(self)
+        d["petition_decision_data_bag"] = [
+            dec.to_dict() for dec in self.petition_decision_data_bag
+        ]
+        return {
+            to_camel_case(k): v
+            for k, v in d.items()
+            if v is not None and (not isinstance(v, list) or v)
+        }
+

tests/clients/test_petition_decisions_client.py

@@ -0,0 +1,27 @@
+"""Tests for PetitionDecisionsClient."""
+
+from typing import Any, Dict
+from unittest.mock import MagicMock, patch
+
+from pyUSPTO.clients.petition_decisions import PetitionDecisionsClient
+from pyUSPTO.models.petition_decisions import PetitionDecisionsResponse
+
+
+class TestPetitionDecisionsClient:
+    def test_search_decisions(self, mock_petition_decisions_client: PetitionDecisionsClient, petition_decisions_sample: Dict[str, Any]) -> None:
+        mock_response = PetitionDecisionsResponse.from_dict(petition_decisions_sample)
+
+        with patch.object(
+            mock_petition_decisions_client,
+            "_make_request",
+            return_value=mock_response,
+        ) as mock_request:
+            result = mock_petition_decisions_client.search_decisions(query="test", limit=5)
+
+        mock_request.assert_called_once_with(
+            method="GET",
+            endpoint="api/v1/petition/decisions/search",
+            params={"q": "test", "limit": 5},
+            response_class=PetitionDecisionsResponse,
+        )
+        assert result is mock_response