From 6d4088bbe30176f99b29094791f2e175a6e34d1f Mon Sep 17 00:00:00 2001 From: Andrew Date: Fri, 12 Dec 2025 10:01:26 -0600 Subject: [PATCH 1/4] implement session sharing --- README.md | 2 +- src/pyUSPTO/clients/base.py | 99 ++++++++++++++++++++--- src/pyUSPTO/clients/patent_data.py | 4 +- src/pyUSPTO/clients/petition_decisions.py | 6 +- src/pyUSPTO/clients/ptab_appeals.py | 4 +- src/pyUSPTO/clients/ptab_interferences.py | 4 +- src/pyUSPTO/clients/ptab_trials.py | 4 +- src/pyUSPTO/config.py | 7 ++ 8 files changed, 111 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index 1c93681..f889735 100644 --- a/README.md +++ b/README.md @@ -124,7 +124,7 @@ interferences_client = PTABInterferencesClient(config=config) ```python # Search for applications by inventor name inventor_search = patent_client.search_applications(inventor_name_q="Smith") -print(f"Found {inventor_search.count} applications with 'Smith' as inventor") +print(f"Found {inventor_search.count} applications with 'Smith' as inventor.") # > Found 104926 applications with 'Smith' as inventor. ``` diff --git a/src/pyUSPTO/clients/base.py b/src/pyUSPTO/clients/base.py index 364405e..2ed1bd8 100644 --- a/src/pyUSPTO/clients/base.py +++ b/src/pyUSPTO/clients/base.py @@ -55,7 +55,9 @@ def __init__( Args: api_key: API key for authentication base_url: The base URL of the API - config: Optional USPTOConfig instance + config: Optional USPTOConfig instance. When multiple clients share the same + config object, they automatically share an HTTP session for better + performance and connection pooling. """ # Handle config if provided if config: @@ -71,32 +73,53 @@ def __init__( # Extract HTTP config for session creation self.http_config = self.config.http_config - # Create session with HTTP config settings - self.session = self._create_session() + # Use shared session from config if available, otherwise create new one + if self.config._shared_session is not None: + # Reuse existing shared session + self.session = self.config._shared_session + self._owns_session = False + # Still apply API key headers in case this client has a different key + self._apply_session_headers() + else: + # Create new session and store in config for sharing + self.session = self._create_session() + self.config._shared_session = self.session + self._owns_session = True - def _create_session(self) -> requests.Session: - """Create configured HTTP session from HTTPConfig settings. + def _apply_session_headers(self) -> None: + """Apply API key and custom headers to the session. - Returns: - Configured requests.Session instance + This is separated from _create_session so it can be used when + a session is injected from outside. """ - session = requests.Session() - # Set API key and default headers if self._api_key: - session.headers.update( + self.session.headers.update( {"X-API-KEY": self._api_key, "content-type": "application/json"} ) # Apply custom headers from HTTP config if self.http_config.custom_headers: - session.headers.update(self.http_config.custom_headers) + self.session.headers.update(self.http_config.custom_headers) + + def _create_session(self) -> requests.Session: + """Create configured HTTP session from HTTPConfig settings. + + Returns: + Configured requests.Session instance + """ + session = requests.Session() + self.session = session + + # Apply headers using shared helper + self._apply_session_headers() # Configure retry strategy from HTTP config retry_strategy = Retry( total=self.http_config.max_retries, backoff_factor=self.http_config.backoff_factor, status_forcelist=self.http_config.retry_status_codes, + allowed_methods=["GET", "POST", "HEAD", "PUT", "DELETE", "OPTIONS", "TRACE"], ) # Create adapter with retry and connection pool settings @@ -111,6 +134,60 @@ def _create_session(self) -> requests.Session: return session + def close(self) -> None: + """Close the HTTP session and release connection pool resources. + + This method should be called when you're done using the client to ensure + proper cleanup of connection pools and resources. Alternatively, use the + client as a context manager for automatic cleanup. + + Note: If a session was provided via the `session` parameter during + initialization, this method will NOT close it, as the client does not + own the session lifecycle. Only sessions created by the client are closed. + + Example: + client = PatentDataClient(api_key="...") + try: + # Use client + pass + finally: + client.close() + """ + if hasattr(self, '_owns_session') and self._owns_session: + if hasattr(self, 'session') and self.session: + self.session.close() + elif not hasattr(self, '_owns_session'): + # Backward compatibility: if _owns_session not set, close anyway + if hasattr(self, 'session') and self.session: + self.session.close() + + def __enter__(self) -> "BaseUSPTOClient[T]": + """Enter context manager, returning the client instance. + + Returns: + Self for use in with statements + + Example: + with PatentDataClient(api_key="...") as client: + response = client.search_applications(...) + """ + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_val: BaseException | None, + exc_tb: Any | None, + ) -> None: + """Exit context manager, ensuring session cleanup. + + Args: + exc_type: Exception type if an exception occurred + exc_val: Exception value if an exception occurred + exc_tb: Exception traceback if an exception occurred + """ + self.close() + def _make_request( self, method: str, diff --git a/src/pyUSPTO/clients/patent_data.py b/src/pyUSPTO/clients/patent_data.py index 3389421..a95cfa4 100644 --- a/src/pyUSPTO/clients/patent_data.py +++ b/src/pyUSPTO/clients/patent_data.py @@ -69,7 +69,9 @@ def __init__( base_url or self.config.patent_data_base_url or "https://api.uspto.gov" ) super().__init__( - api_key=api_key_to_use, base_url=effective_base_url, config=self.config + api_key=api_key_to_use, + base_url=effective_base_url, + config=self.config, ) def sanitize_application_number(self, input_number: str) -> str: diff --git a/src/pyUSPTO/clients/petition_decisions.py b/src/pyUSPTO/clients/petition_decisions.py index a9b8ca2..1507797 100644 --- a/src/pyUSPTO/clients/petition_decisions.py +++ b/src/pyUSPTO/clients/petition_decisions.py @@ -10,8 +10,6 @@ from pathlib import Path from typing import Any -import requests - from pyUSPTO.clients.base import BaseUSPTOClient from pyUSPTO.config import USPTOConfig from pyUSPTO.models.petition_decisions import ( @@ -60,7 +58,9 @@ def __init__( or "https://api.uspto.gov" ) super().__init__( - api_key=api_key_to_use, base_url=effective_base_url, config=self.config + api_key=api_key_to_use, + base_url=effective_base_url, + config=self.config, ) def _get_decision_from_response( diff --git a/src/pyUSPTO/clients/ptab_appeals.py b/src/pyUSPTO/clients/ptab_appeals.py index 7ef848f..d75fc5d 100644 --- a/src/pyUSPTO/clients/ptab_appeals.py +++ b/src/pyUSPTO/clients/ptab_appeals.py @@ -45,7 +45,9 @@ def __init__( base_url or self.config.ptab_base_url or "https://api.uspto.gov" ) super().__init__( - api_key=api_key_to_use, base_url=effective_base_url, config=self.config + api_key=api_key_to_use, + base_url=effective_base_url, + config=self.config, ) def search_decisions( diff --git a/src/pyUSPTO/clients/ptab_interferences.py b/src/pyUSPTO/clients/ptab_interferences.py index ff55314..007ad5a 100644 --- a/src/pyUSPTO/clients/ptab_interferences.py +++ b/src/pyUSPTO/clients/ptab_interferences.py @@ -45,7 +45,9 @@ def __init__( base_url or self.config.ptab_base_url or "https://api.uspto.gov" ) super().__init__( - api_key=api_key_to_use, base_url=effective_base_url, config=self.config + api_key=api_key_to_use, + base_url=effective_base_url, + config=self.config, ) def search_decisions( diff --git a/src/pyUSPTO/clients/ptab_trials.py b/src/pyUSPTO/clients/ptab_trials.py index bc2bc57..897f2fa 100644 --- a/src/pyUSPTO/clients/ptab_trials.py +++ b/src/pyUSPTO/clients/ptab_trials.py @@ -54,7 +54,9 @@ def __init__( base_url or self.config.ptab_base_url or "https://api.uspto.gov" ) super().__init__( - api_key=api_key_to_use, base_url=effective_base_url, config=self.config + api_key=api_key_to_use, + base_url=effective_base_url, + config=self.config, ) def _perform_search( diff --git a/src/pyUSPTO/config.py b/src/pyUSPTO/config.py index 72fedbc..0e3e480 100644 --- a/src/pyUSPTO/config.py +++ b/src/pyUSPTO/config.py @@ -5,6 +5,10 @@ """ import os +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + import requests from pyUSPTO.http_config import HTTPConfig @@ -52,6 +56,9 @@ def __init__( # Control whether to include raw JSON data in response objects self.include_raw_data = include_raw_data + # Shared session for all clients using this config (created lazily) + self._shared_session: requests.Session | None = None + @classmethod def from_env(cls) -> "USPTOConfig": """Create a USPTOConfig from environment variables. From 548df758d827325d0ca57c3054fd6fa2e9b6d690 Mon Sep 17 00:00:00 2001 From: Andrew Date: Fri, 12 Dec 2025 10:03:03 -0600 Subject: [PATCH 2/4] Configurable Download Chunk Size --- src/pyUSPTO/clients/base.py | 2 +- src/pyUSPTO/clients/bulk_data.py | 2 +- src/pyUSPTO/http_config.py | 21 +++++++++++++++++++++ 3 files changed, 23 insertions(+), 2 deletions(-) diff --git a/src/pyUSPTO/clients/base.py b/src/pyUSPTO/clients/base.py index 2ed1bd8..0a66454 100644 --- a/src/pyUSPTO/clients/base.py +++ b/src/pyUSPTO/clients/base.py @@ -473,7 +473,7 @@ def _save_response_to_file( # Save to disk with streaming with open(file=str(path), mode="wb") as f: - for chunk in response.iter_content(chunk_size=8192): + for chunk in response.iter_content(chunk_size=self.http_config.download_chunk_size): if chunk: # Filter out keep-alive chunks f.write(chunk) return str(path) diff --git a/src/pyUSPTO/clients/bulk_data.py b/src/pyUSPTO/clients/bulk_data.py index 7d140c8..f89d8e8 100644 --- a/src/pyUSPTO/clients/bulk_data.py +++ b/src/pyUSPTO/clients/bulk_data.py @@ -192,7 +192,7 @@ def download_file(self, file_data: FileData, destination: str) -> str: file_path = os.path.join(destination, file_data.file_name) with open(file_path, "wb") as f: - for chunk in result.iter_content(chunk_size=8192): + for chunk in result.iter_content(chunk_size=self.http_config.download_chunk_size): f.write(chunk) return file_path diff --git a/src/pyUSPTO/http_config.py b/src/pyUSPTO/http_config.py index 3d0e3c3..64c02b0 100644 --- a/src/pyUSPTO/http_config.py +++ b/src/pyUSPTO/http_config.py @@ -23,6 +23,7 @@ class HTTPConfig: retry_status_codes: HTTP status codes that trigger retries pool_connections: Number of connection pools to cache (default: 10) pool_maxsize: Maximum number of connections per pool (default: 10) + download_chunk_size: Chunk size in bytes for streaming file downloads (default: 8192) custom_headers: Additional headers to include in all requests """ @@ -41,9 +42,27 @@ class HTTPConfig: pool_connections: int = 10 pool_maxsize: int = 10 + # Download configuration + download_chunk_size: int = 8192 # Bytes per chunk when streaming downloads + # Custom headers (User-Agent, tracking, etc.) custom_headers: dict[str, str] | None = None + def __post_init__(self) -> None: + """Validate configuration after initialization.""" + if self.download_chunk_size <= 0: + raise ValueError( + f"download_chunk_size must be positive, got {self.download_chunk_size}" + ) + if self.download_chunk_size > 10485760: # 10 MB + import warnings + + warnings.warn( + f"download_chunk_size of {self.download_chunk_size} bytes is very large " + "and may cause memory issues. Consider using <= 1048576 (1 MB).", + UserWarning, + ) + @classmethod def from_env(cls) -> "HTTPConfig": """Create HTTPConfig from environment variables. @@ -55,6 +74,7 @@ def from_env(cls) -> "HTTPConfig": USPTO_BACKOFF_FACTOR: Retry backoff factor USPTO_POOL_CONNECTIONS: Connection pool size USPTO_POOL_MAXSIZE: Max connections per pool + USPTO_DOWNLOAD_CHUNK_SIZE: Chunk size for streaming downloads (bytes) Returns: HTTPConfig instance with values from environment or defaults @@ -66,6 +86,7 @@ def from_env(cls) -> "HTTPConfig": backoff_factor=float(os.environ.get("USPTO_BACKOFF_FACTOR", "1.0")), pool_connections=int(os.environ.get("USPTO_POOL_CONNECTIONS", "10")), pool_maxsize=int(os.environ.get("USPTO_POOL_MAXSIZE", "10")), + download_chunk_size=int(os.environ.get("USPTO_DOWNLOAD_CHUNK_SIZE", "8192")), ) def get_timeout_tuple(self) -> tuple[float | None, float | None]: From 0cd005710a18164c9c1e9573a98f51581af82972 Mon Sep 17 00:00:00 2001 From: Andrew Date: Fri, 12 Dec 2025 12:21:47 -0600 Subject: [PATCH 3/4] improve paginate_decisions with POST support --- src/pyUSPTO/clients/base.py | 119 ++++++++++-- src/pyUSPTO/clients/bulk_data.py | 14 +- src/pyUSPTO/clients/patent_data.py | 51 +++--- src/pyUSPTO/clients/petition_decisions.py | 39 ++-- src/pyUSPTO/clients/ptab_appeals.py | 41 +++-- src/pyUSPTO/clients/ptab_interferences.py | 41 +++-- src/pyUSPTO/clients/ptab_trials.py | 21 ++- src/pyUSPTO/http_config.py | 7 +- tests/clients/test_base.py | 169 ++++++++++++++++++ tests/clients/test_bulk_data_clients.py | 7 +- tests/clients/test_patent_data_clients.py | 9 +- .../clients/test_petition_decision_clients.py | 8 +- tests/clients/test_ptab_appeals_client.py | 8 +- .../clients/test_ptab_interferences_client.py | 8 +- tests/clients/test_ptab_trials_client.py | 8 +- 15 files changed, 419 insertions(+), 131 deletions(-) diff --git a/src/pyUSPTO/clients/base.py b/src/pyUSPTO/clients/base.py index 0a66454..ff0cc03 100644 --- a/src/pyUSPTO/clients/base.py +++ b/src/pyUSPTO/clients/base.py @@ -25,6 +25,7 @@ USPTOTimeout, get_api_exception, ) +from pyUSPTO.http_config import ALLOWED_METHODS @runtime_checkable @@ -119,7 +120,7 @@ def _create_session(self) -> requests.Session: total=self.http_config.max_retries, backoff_factor=self.http_config.backoff_factor, status_forcelist=self.http_config.retry_status_codes, - allowed_methods=["GET", "POST", "HEAD", "PUT", "DELETE", "OPTIONS", "TRACE"], + allowed_methods=ALLOWED_METHODS, ) # Create adapter with retry and connection pool settings @@ -153,12 +154,12 @@ def close(self) -> None: finally: client.close() """ - if hasattr(self, '_owns_session') and self._owns_session: - if hasattr(self, 'session') and self.session: + if hasattr(self, "_owns_session") and self._owns_session: + if hasattr(self, "session") and self.session: self.session.close() - elif not hasattr(self, '_owns_session'): + elif not hasattr(self, "_owns_session"): # Backward compatibility: if _owns_session not set, close anyway - if hasattr(self, 'session') and self.session: + if hasattr(self, "session") and self.session: self.session.close() def __enter__(self) -> "BaseUSPTOClient[T]": @@ -263,6 +264,14 @@ def _make_request( except requests.exceptions.HTTPError as http_err: client_operation_message = f"API request to '{url}' failed with HTTPError" # 'url' is from _make_request scope + # Include request body for POST debugging + if method.upper() == "POST" and json_data: + import json + + client_operation_message += ( + f"\nRequest body sent:\n{json.dumps(json_data, indent=2)}" + ) + # Create APIErrorArgs directly from the HTTPError current_error_args = APIErrorArgs.from_http_error( http_error=http_err, client_operation_message=client_operation_message @@ -306,27 +315,103 @@ def _make_request( raise api_exception_to_raise from req_err def paginate_results( - self, method_name: str, response_container_attr: str, **kwargs: Any + self, + method_name: str, + response_container_attr: str, + post_body: dict[str, Any] | None = None, + **kwargs: Any, ) -> Generator[Any, None, None]: - """Paginate through all results of a method. + """Paginate through all results of a method, supporting both GET and POST. Args: method_name: Name of the method to call response_container_attr: Attribute name of the container in the response - **kwargs: Keyword arguments to pass to the method + post_body: Optional POST body for POST-based pagination. If provided, + pagination parameters (offset, limit) will be injected into this body. + **kwargs: Keyword arguments to pass to the method (for GET pagination) Yields: Items from the response container + + Raises: + ValueError: If offset is provided in kwargs or post_body (offset is managed + automatically by pagination) + + Examples: + # GET pagination + for app in client.paginate_results( + "search_applications", + "patent_file_wrapper_data_bag", + query="test" + ): + print(app) + + # POST pagination with custom limit + for app in client.paginate_results( + "search_applications", + "patent_file_wrapper_data_bag", + post_body={"q": "test", "limit": 50} + ): + print(app) """ - offset = kwargs.get("offset", 0) - limit = kwargs.get("limit", 25) + # Determine if POST body uses nested pagination structure + uses_nested_pagination = False + if post_body is not None: + uses_nested_pagination = "pagination" in post_body and isinstance( + post_body["pagination"], dict + ) + + # Validate that offset is not provided by the user + if post_body is not None: + if uses_nested_pagination: + # Check nested pagination object + if "offset" in post_body["pagination"]: + raise ValueError( + "Cannot specify 'offset' in post_body['pagination']. " + "Pagination manages offset automatically." + ) + limit = post_body["pagination"].get("limit", 25) + else: + # Check top-level + if "offset" in post_body: + raise ValueError( + "Cannot specify 'offset' in post_body. Pagination manages offset automatically." + ) + limit = post_body.get("limit", 25) + else: + if "offset" in kwargs: + raise ValueError( + "Cannot specify 'offset' in kwargs. Pagination manages offset automatically." + ) + limit = kwargs.get("limit", 25) + + offset = 0 while True: - kwargs["offset"] = offset - kwargs["limit"] = limit + # Prepare parameters based on request type + if post_body is not None: + # POST request: update body with pagination params + current_body = post_body.copy() + + if uses_nested_pagination: + # Update nested pagination object + current_body["pagination"] = current_body["pagination"].copy() + current_body["pagination"]["offset"] = offset + current_body["pagination"]["limit"] = limit + else: + # Update top-level pagination params + current_body["offset"] = offset + current_body["limit"] = limit + + method = getattr(self, method_name) + response = method(post_body=current_body, **kwargs) + else: + # GET request: update kwargs with pagination params + kwargs["offset"] = offset + kwargs["limit"] = limit - method = getattr(self, method_name) - response = method(**kwargs) + method = getattr(self, method_name) + response = method(**kwargs) if not response.count: break @@ -334,7 +419,7 @@ def paginate_results( container = getattr(response, response_container_attr) yield from container - if response.count < limit: + if response.count < limit + offset: break offset += limit @@ -473,7 +558,9 @@ def _save_response_to_file( # Save to disk with streaming with open(file=str(path), mode="wb") as f: - for chunk in response.iter_content(chunk_size=self.http_config.download_chunk_size): + for chunk in response.iter_content( + chunk_size=self.http_config.download_chunk_size + ): if chunk: # Filter out keep-alive chunks f.write(chunk) return str(path) diff --git a/src/pyUSPTO/clients/bulk_data.py b/src/pyUSPTO/clients/bulk_data.py index f89d8e8..b196e60 100644 --- a/src/pyUSPTO/clients/bulk_data.py +++ b/src/pyUSPTO/clients/bulk_data.py @@ -192,16 +192,23 @@ def download_file(self, file_data: FileData, destination: str) -> str: file_path = os.path.join(destination, file_data.file_name) with open(file_path, "wb") as f: - for chunk in result.iter_content(chunk_size=self.http_config.download_chunk_size): + for chunk in result.iter_content( + chunk_size=self.http_config.download_chunk_size + ): f.write(chunk) return file_path - def paginate_products(self, **kwargs: Any) -> Iterator[BulkDataProduct]: + def paginate_products( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Iterator[BulkDataProduct]: """Paginate through all products matching the search criteria. + Supports both GET and POST requests. + Args: - **kwargs: Keyword arguments to pass to search_products + post_body: Optional POST body for complex search queries + **kwargs: Keyword arguments for GET-based pagination Yields: BulkDataProduct objects @@ -209,6 +216,7 @@ def paginate_products(self, **kwargs: Any) -> Iterator[BulkDataProduct]: return self.paginate_results( method_name="search_products", response_container_attr="bulk_data_product_bag", + post_body=post_body, **kwargs, ) diff --git a/src/pyUSPTO/clients/patent_data.py b/src/pyUSPTO/clients/patent_data.py index a95cfa4..78868a7 100644 --- a/src/pyUSPTO/clients/patent_data.py +++ b/src/pyUSPTO/clients/patent_data.py @@ -819,45 +819,54 @@ def get_application_associated_documents( wrapper = self._get_wrapper_from_response(response_data, application_number) return PrintedPublication.from_wrapper(wrapper) if wrapper else None - def paginate_applications(self, **kwargs: Any) -> Iterator[PatentFileWrapper]: + def paginate_applications( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Iterator[PatentFileWrapper]: """Provide an iterator to easily paginate through patent application search results. This method simplifies the process of fetching all patent applications that match a given search query by automatically handling pagination. - It internally calls the `search_applications` method for GET requests, - batching results and yielding them one by one. + Supports both GET and POST requests. + + For GET requests, provide search parameters as keyword arguments. + For POST requests, provide the search criteria in `post_body`. - All keyword arguments provided to this method (`**kwargs`) are passed - directly to the `search_applications` method to define the search - criteria. See the `search_applications` method for more details - on available parameters. The `offset` and `limit` parameters are managed by the pagination logic; - setting them directly in `kwargs` might lead to unexpected behavior. + setting them directly in `kwargs` or `post_body` might lead to unexpected behavior. Args: - **kwargs (Any): Keyword arguments to be passed to the - `search_applications` method for constructing the search query. - These define the criteria for the patent applications to be - retrieved. Do not include `post_body`. + post_body: Optional POST body for complex search queries. If provided, + performs POST-based pagination. + **kwargs: Keyword arguments for GET-based pagination or additional + query parameters for POST requests. Returns: Iterator[PatentFileWrapper]: An iterator that yields `PatentFileWrapper` objects, allowing iteration over all matching patent applications across multiple pages of results. - Raises: - ValueError: If `post_body` is included in `kwargs`, as this - method only supports GET request parameters for pagination. + Examples: + # GET-based pagination + for wrapper in client.paginate_applications( + query="applicationNumberText:16*", + limit=50 + ): + print(wrapper.application_number_text) + + # POST-based pagination + for wrapper in client.paginate_applications( + post_body={ + "q": "applicationNumberText:16*", + "facets": "true", + "fields": "applicationNumberText,applicationMetaData" + } + ): + print(wrapper.application_number_text) """ - if "post_body" in kwargs: - raise ValueError( - "paginate_applications uses GET requests and does not support 'post_body'. " - "Use keyword arguments for search criteria." - ) - return self.paginate_results( method_name="search_applications", response_container_attr="patent_file_wrapper_data_bag", + post_body=post_body, **kwargs, ) diff --git a/src/pyUSPTO/clients/petition_decisions.py b/src/pyUSPTO/clients/petition_decisions.py index 1507797..786b8cd 100644 --- a/src/pyUSPTO/clients/petition_decisions.py +++ b/src/pyUSPTO/clients/petition_decisions.py @@ -10,6 +10,8 @@ from pathlib import Path from typing import Any +import requests + from pyUSPTO.clients.base import BaseUSPTOClient from pyUSPTO.config import USPTOConfig from pyUSPTO.models.petition_decisions import ( @@ -478,51 +480,44 @@ def download_decisions( # Return streaming response for manual handling return result - def paginate_decisions(self, **kwargs: Any) -> Iterator[PetitionDecision]: + def paginate_decisions( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Iterator[PetitionDecision]: """Provide an iterator to paginate through petition decision search results. This method simplifies fetching all petition decisions matching a search query - by automatically handling pagination. It internally calls the search_decisions - method for GET requests, batching results and yielding them one by one. + by automatically handling pagination. Supports both GET and POST requests. - All keyword arguments are passed directly to search_decisions to define the - search criteria. The offset and limit parameters are managed by the pagination - logic; setting them directly in kwargs might lead to unexpected behavior. + The offset and limit parameters are managed by the pagination logic; + setting them directly in kwargs or post_body might lead to unexpected behavior. Args: - **kwargs: Keyword arguments passed to search_decisions for constructing - the search query. Do not include post_body. + post_body: Optional POST body for complex search queries + **kwargs: Keyword arguments for GET-based pagination Returns: Iterator[PetitionDecision]: An iterator yielding PetitionDecision objects, allowing iteration over all matching petition decisions across multiple pages of results. - Raises: - ValueError: If post_body is included in kwargs, as this method only - supports GET request parameters for pagination. - Examples: - # Paginate through all decisions for a technology center + # GET pagination through all decisions for a technology center >>> for decision in client.paginate_decisions(technology_center_q="1700"): ... print(f"{decision.application_number_text}: {decision.decision_type_code}") - # Paginate with date range + # POST pagination with date range >>> for decision in client.paginate_decisions( - ... decision_date_from_q="2023-01-01", - ... decision_date_to_q="2023-12-31" + ... post_body={ + ... "decision_date_from_q": "2023-01-01", + ... "decision_date_to_q": "2023-12-31" + ... } ... ): ... process_decision(decision) """ - if "post_body" in kwargs: - raise ValueError( - "paginate_decisions uses GET requests and does not support 'post_body'. " - "Use keyword arguments for search criteria." - ) - return self.paginate_results( method_name="search_decisions", response_container_attr="petition_decision_data_bag", + post_body=post_body, **kwargs, ) diff --git a/src/pyUSPTO/clients/ptab_appeals.py b/src/pyUSPTO/clients/ptab_appeals.py index d75fc5d..4796e9b 100644 --- a/src/pyUSPTO/clients/ptab_appeals.py +++ b/src/pyUSPTO/clients/ptab_appeals.py @@ -206,50 +206,53 @@ def search_decisions( assert isinstance(result, PTABAppealResponse) return result - def paginate_decisions(self, **kwargs: Any) -> Iterator[PTABAppealDecision]: + def paginate_decisions( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Iterator[PTABAppealDecision]: """Provide an iterator to paginate through appeal decision search results. This method simplifies fetching all appeal decisions matching a search query by automatically handling pagination. It internally calls the search_decisions - method for GET requests, batching results and yielding them one by one. + method, batching results and yielding them one by one. - All keyword arguments are passed directly to search_decisions to define the - search criteria. The offset and limit parameters are managed by the pagination - logic; setting them directly in kwargs might lead to unexpected behavior. + Supports both GET and POST requests. For POST requests, provide the + search criteria in `post_body`. For GET requests, use keyword arguments. + + The offset parameter is managed by the pagination logic and should not be + provided by the user. The limit parameter can be customized. Args: + post_body: Optional POST body for complex search queries. **kwargs: Keyword arguments passed to search_decisions for constructing - the search query. Do not include post_body. + the search query (for GET-based pagination). Returns: Iterator[PTABAppealDecision]: An iterator yielding PTABAppealDecision objects, allowing iteration over all matching decisions across multiple pages of results. - Raises: - ValueError: If post_body is included in kwargs, as this method only - supports GET request parameters for pagination. - Examples: - # Paginate through all decisions for a technology center + # GET-based pagination with convenience parameters >>> for decision in client.paginate_decisions(technology_center_number_q="3600"): ... print(f"{decision.appeal_meta_data.appeal_number}: " ... f"{decision.decision_data.decision_type_category}") - # Paginate with date range + # GET-based pagination with date range and custom limit >>> for decision in client.paginate_decisions( ... decision_date_from_q="2023-01-01", - ... decision_date_to_q="2023-12-31" + ... decision_date_to_q="2023-12-31", + ... limit=50 ... ): ... process_decision(decision) - """ - if "post_body" in kwargs: - raise ValueError( - "paginate_decisions uses GET requests and does not support 'post_body'. " - "Use keyword arguments for search criteria." - ) + # POST-based pagination + >>> for decision in client.paginate_decisions( + ... post_body={"q": "decisionTypeCategory:Affirmed", "limit": 100} + ... ): + ... process_decision(decision) + """ return self.paginate_results( method_name="search_decisions", response_container_attr="patent_appeal_data_bag", + post_body=post_body, **kwargs, ) diff --git a/src/pyUSPTO/clients/ptab_interferences.py b/src/pyUSPTO/clients/ptab_interferences.py index 007ad5a..5ad272d 100644 --- a/src/pyUSPTO/clients/ptab_interferences.py +++ b/src/pyUSPTO/clients/ptab_interferences.py @@ -221,51 +221,54 @@ def search_decisions( assert isinstance(result, PTABInterferenceResponse) return result - def paginate_decisions(self, **kwargs: Any) -> Iterator[PTABInterferenceDecision]: + def paginate_decisions( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Iterator[PTABInterferenceDecision]: """Provide an iterator to paginate through interference decision search results. This method simplifies fetching all interference decisions matching a search query by automatically handling pagination. It internally calls the search_decisions - method for GET requests, batching results and yielding them one by one. + method, batching results and yielding them one by one. - All keyword arguments are passed directly to search_decisions to define the - search criteria. The offset and limit parameters are managed by the pagination - logic; setting them directly in kwargs might lead to unexpected behavior. + Supports both GET and POST requests. For POST requests, provide the + search criteria in `post_body`. For GET requests, use keyword arguments. + + The offset parameter is managed by the pagination logic and should not be + provided by the user. The limit parameter can be customized. Args: + post_body: Optional POST body for complex search queries. **kwargs: Keyword arguments passed to search_decisions for constructing - the search query. Do not include post_body. + the search query (for GET-based pagination). Returns: Iterator[PTABInterferenceDecision]: An iterator yielding PTABInterferenceDecision objects, allowing iteration over all matching decisions across multiple pages of results. - Raises: - ValueError: If post_body is included in kwargs, as this method only - supports GET request parameters for pagination. - Examples: - # Paginate through all interference decisions + # GET-based pagination through all decisions >>> for decision in client.paginate_decisions(): ... print(f"{decision.interference_meta_data.interference_number}: " ... f"{decision.document_data.interference_outcome_category}") - # Paginate with date range + # GET-based pagination with date range and custom limit >>> for decision in client.paginate_decisions( ... decision_date_from_q="2020-01-01", - ... decision_date_to_q="2023-12-31" + ... decision_date_to_q="2023-12-31", + ... limit=50 ... ): ... process_decision(decision) - """ - if "post_body" in kwargs: - raise ValueError( - "paginate_decisions uses GET requests and does not support 'post_body'. " - "Use keyword arguments for search criteria." - ) + # POST-based pagination + >>> for decision in client.paginate_decisions( + ... post_body={"q": "interferenceOutcomeCategory:Priority to Senior Party"} + ... ): + ... process_decision(decision) + """ return self.paginate_results( method_name="search_decisions", response_container_attr="patent_interference_data_bag", + post_body=post_body, **kwargs, ) diff --git a/src/pyUSPTO/clients/ptab_trials.py b/src/pyUSPTO/clients/ptab_trials.py index 897f2fa..d24802a 100644 --- a/src/pyUSPTO/clients/ptab_trials.py +++ b/src/pyUSPTO/clients/ptab_trials.py @@ -483,16 +483,23 @@ def search_decisions( additional_params=additional_query_params, ) # type: ignore - def paginate_proceedings(self, **kwargs: Any) -> Iterator[PTABTrialProceeding]: - """Provide an iterator to paginate through trial proceeding search results.""" - if "post_body" in kwargs: - raise ValueError( - "paginate_proceedings uses GET requests and does not support 'post_body'." - "Use keyword arguments for search criteria." - ) + def paginate_proceedings( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Iterator[PTABTrialProceeding]: + """Provide an iterator to paginate through trial proceeding search results. + + Supports both GET and POST requests. + + Args: + post_body: Optional POST body for complex search queries + **kwargs: Keyword arguments for GET-based pagination + Yields: + PTABTrialProceeding objects + """ return self.paginate_results( method_name="search_proceedings", response_container_attr="patent_trial_proceeding_data_bag", + post_body=post_body, **kwargs, ) diff --git a/src/pyUSPTO/http_config.py b/src/pyUSPTO/http_config.py index 64c02b0..b7f311c 100644 --- a/src/pyUSPTO/http_config.py +++ b/src/pyUSPTO/http_config.py @@ -7,6 +7,9 @@ import os from dataclasses import dataclass, field +# HTTP methods supported by the USPTO API +ALLOWED_METHODS = ["GET", "POST"] + @dataclass class HTTPConfig: @@ -86,7 +89,9 @@ def from_env(cls) -> "HTTPConfig": backoff_factor=float(os.environ.get("USPTO_BACKOFF_FACTOR", "1.0")), pool_connections=int(os.environ.get("USPTO_POOL_CONNECTIONS", "10")), pool_maxsize=int(os.environ.get("USPTO_POOL_MAXSIZE", "10")), - download_chunk_size=int(os.environ.get("USPTO_DOWNLOAD_CHUNK_SIZE", "8192")), + download_chunk_size=int( + os.environ.get("USPTO_DOWNLOAD_CHUNK_SIZE", "8192") + ), ) def get_timeout_tuple(self) -> tuple[float | None, float | None]: diff --git a/tests/clients/test_base.py b/tests/clients/test_base.py index 6625aa8..69eaa86 100644 --- a/tests/clients/test_base.py +++ b/tests/clients/test_base.py @@ -562,6 +562,175 @@ def test_method(self, **kwargs: Any) -> Any: # Verify empty results works assert results == [] + def test_paginate_results_with_nested_pagination( + self, mock_session: MagicMock + ) -> None: + """Test paginate_results handles nested pagination structure correctly.""" + # Create mock responses + first_response = MagicMock() + first_response.count = 2 + first_response.items = ["item1", "item2"] + + second_response = MagicMock() + second_response.count = 1 + second_response.items = ["item3"] + + third_response = MagicMock() + third_response.count = 0 + third_response.items = [] + + # Track what post_body was actually sent + received_bodies: list[dict[str, Any]] = [] + + class TestClient(BaseUSPTOClient[Any]): + def test_method( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Any: + # Record the body we received + if post_body: + received_bodies.append(post_body.copy()) + + # Return different responses based on offset in nested pagination + if post_body and "pagination" in post_body: + offset = post_body["pagination"].get("offset", 0) + if offset == 0: + return first_response + elif offset == 2: + return second_response + else: + return third_response + return third_response + + test_client = TestClient(base_url="https://api.test.com") + test_client.session = mock_session + + # Test with nested pagination structure (like USPTO API accepts) + post_body = { + "q": "applicationMetaData.applicationStatusDate:>2025-06-16", + "filters": [{"name": "eventDataBag.eventCode", "value": ["CTNF", "CTFR"]}], + "fields": ["applicationNumberText", "eventDataBag"], + "pagination": {"limit": 2}, # Nested structure + } + + results = list( + test_client.paginate_results( + method_name="test_method", + response_container_attr="items", + post_body=post_body, + ) + ) + + # Verify results + assert results == ["item1", "item2", "item3"] + + # Verify that pagination params were added to nested structure, not top-level + assert len(received_bodies) == 2 + + # First request should have offset=0, limit=2 in nested pagination + first_body = received_bodies[0] + assert "pagination" in first_body + assert first_body["pagination"]["offset"] == 0 + assert first_body["pagination"]["limit"] == 2 + # Should NOT have top-level offset/limit + assert "offset" not in first_body + assert "limit" not in first_body + # Original fields should be preserved + assert first_body["q"] == "applicationMetaData.applicationStatusDate:>2025-06-16" + assert len(first_body["filters"]) == 1 + assert first_body["fields"] == ["applicationNumberText", "eventDataBag"] + + # Second request should have offset=2, limit=2 in nested pagination + second_body = received_bodies[1] + assert "pagination" in second_body + assert second_body["pagination"]["offset"] == 2 + assert second_body["pagination"]["limit"] == 2 + # Should NOT have top-level offset/limit + assert "offset" not in second_body + assert "limit" not in second_body + + def test_paginate_results_with_flat_pagination( + self, mock_session: MagicMock + ) -> None: + """Test paginate_results still works with flat (top-level) pagination structure.""" + # Create mock responses + first_response = MagicMock() + first_response.count = 2 + first_response.items = ["item1", "item2"] + + second_response = MagicMock() + second_response.count = 0 + second_response.items = [] + + received_bodies: list[dict[str, Any]] = [] + + class TestClient(BaseUSPTOClient[Any]): + def test_method( + self, post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Any: + if post_body: + received_bodies.append(post_body.copy()) + + # Return different responses based on top-level offset + if post_body: + offset = post_body.get("offset", 0) + if offset == 0: + return first_response + else: + return second_response + return second_response + + test_client = TestClient(base_url="https://api.test.com") + test_client.session = mock_session + + # Test with flat (top-level) pagination structure + post_body = { + "q": "test query", + "limit": 2, # Flat structure + } + + results = list( + test_client.paginate_results( + method_name="test_method", + response_container_attr="items", + post_body=post_body, + ) + ) + + # Verify results + assert results == ["item1", "item2"] + + # Verify that pagination params were added at top-level + assert len(received_bodies) == 1 + first_body = received_bodies[0] + assert first_body["offset"] == 0 + assert first_body["limit"] == 2 + # Should NOT have nested pagination + assert "pagination" not in first_body + + def test_paginate_results_rejects_offset_in_nested_pagination( + self, mock_session: MagicMock + ) -> None: + """Test that offset is rejected when provided in nested pagination.""" + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + client.session = mock_session + + post_body = { + "q": "test", + "pagination": {"offset": 10, "limit": 50}, # User provided offset - BAD + } + + with pytest.raises( + ValueError, + match="Cannot specify 'offset' in post_body\\['pagination'\\]", + ): + list( + client.paginate_results( + method_name="test_method", + response_container_attr="items", + post_body=post_body, + ) + ) + def test_save_response_to_file(self, mock_session: MagicMock) -> None: """Test _save_response_to_file raises FileExistsError.""" client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") diff --git a/tests/clients/test_bulk_data_clients.py b/tests/clients/test_bulk_data_clients.py index ddc94d0..9c28530 100644 --- a/tests/clients/test_bulk_data_clients.py +++ b/tests/clients/test_bulk_data_clients.py @@ -268,7 +268,7 @@ def test_get_products( # Verify mock_session.get.assert_called_once_with( - url=f"{mock_bulk_data_client.base_url}/products/search", + url=f"{mock_bulk_data_client.base_url}/api/v1/datasets/products/search", params={"param": "value"}, stream=False, timeout=(10.0, 30.0), @@ -308,7 +308,7 @@ def test_get_product_by_id( # Verify mock_session.get.assert_called_once_with( - url=f"{mock_bulk_data_client.base_url}/products/{product_id}", + url=f"{mock_bulk_data_client.base_url}/api/v1/datasets/products/{product_id}", params={ "fileDataFromDate": "2023-01-01", "fileDataToDate": "2023-12-31", @@ -454,7 +454,7 @@ def test_search_products( # Verify mock_session.get.assert_called_once_with( - url=f"{mock_bulk_data_client.base_url}/products/search", + url=f"{mock_bulk_data_client.base_url}/api/v1/datasets/products/search", params={ "q": "test", "productTitle": "Test Product", @@ -497,6 +497,7 @@ def test_paginate_products(self, mock_bulk_data_client: BulkDataClient) -> None: mock_paginate_results.assert_called_once_with( method_name="search_products", response_container_attr="bulk_data_product_bag", + post_body=None, param="value", ) diff --git a/tests/clients/test_patent_data_clients.py b/tests/clients/test_patent_data_clients.py index 543c3c6..1731e74 100644 --- a/tests/clients/test_patent_data_clients.py +++ b/tests/clients/test_patent_data_clients.py @@ -657,6 +657,7 @@ def test_paginate_applications(self, patent_data_client: PatentDataClient) -> No mock_paginate_results.assert_called_once_with( method_name="search_applications", response_container_attr="patent_file_wrapper_data_bag", + post_body=None, query="Test", limit=20, ) @@ -664,15 +665,15 @@ def test_paginate_applications(self, patent_data_client: PatentDataClient) -> No assert results[0] is patent1 assert results[1] is patent2 - def test_paginate_applications_raises_value_error_for_post_body( + def test_paginate_applications_rejects_offset_in_kwargs( self, patent_data_client: PatentDataClient ) -> None: - """Test paginate_applications raises ValueError if post_body is provided.""" + """Test paginate_applications raises ValueError if offset is provided in kwargs.""" with pytest.raises( ValueError, - match="paginate_applications uses GET requests and does not support 'post_body'", + match="Cannot specify 'offset'.*Pagination manages offset automatically", ): - list(patent_data_client.paginate_applications(post_body={"q": "test"})) + list(patent_data_client.paginate_applications(query="test", offset=10)) class TestPatentApplicationDocumentListing: diff --git a/tests/clients/test_petition_decision_clients.py b/tests/clients/test_petition_decision_clients.py index 4eba4f9..7334b55 100644 --- a/tests/clients/test_petition_decision_clients.py +++ b/tests/clients/test_petition_decision_clients.py @@ -756,12 +756,12 @@ def test_paginate_decisions( assert results[1].application_number_text == "222" assert results[2].application_number_text == "333" - def test_paginate_decisions_rejects_post_body( + def test_paginate_decisions_rejects_offset_in_kwargs( self, petition_client: FinalPetitionDecisionsClient ) -> None: - """Test that pagination raises error with post_body.""" - with pytest.raises(ValueError, match="does not support 'post_body'"): - list(petition_client.paginate_decisions(post_body={"q": "test"})) + """Test that pagination raises error with offset in kwargs.""" + with pytest.raises(ValueError, match="Cannot specify 'offset'"): + list(petition_client.paginate_decisions(query="test", offset=10)) class TestFinalPetitionDecisionsClientDocumentDownload: diff --git a/tests/clients/test_ptab_appeals_client.py b/tests/clients/test_ptab_appeals_client.py index 668a013..5484cdc 100644 --- a/tests/clients/test_ptab_appeals_client.py +++ b/tests/clients/test_ptab_appeals_client.py @@ -371,12 +371,12 @@ def test_paginate_decisions( assert results[2].appeal_number == "2023-001236" assert mock_search.call_count == 2 # Stops when count < limit - def test_paginate_decisions_raises_on_post_body( + def test_paginate_decisions_rejects_offset_in_kwargs( self, mock_ptab_appeals_client: PTABAppealsClient ) -> None: - """Test that paginate_decisions raises ValueError with post_body.""" - with pytest.raises(ValueError, match="does not support 'post_body'"): - list(mock_ptab_appeals_client.paginate_decisions(post_body={"q": "test"})) + """Test that paginate_decisions raises ValueError with offset in kwargs.""" + with pytest.raises(ValueError, match="Cannot specify 'offset'"): + list(mock_ptab_appeals_client.paginate_decisions(query="test", offset=10)) def test_paginate_decisions_with_multiple_params( self, mock_ptab_appeals_client: PTABAppealsClient diff --git a/tests/clients/test_ptab_interferences_client.py b/tests/clients/test_ptab_interferences_client.py index 4addd38..09fdd90 100644 --- a/tests/clients/test_ptab_interferences_client.py +++ b/tests/clients/test_ptab_interferences_client.py @@ -465,14 +465,14 @@ def test_paginate_decisions( assert results[2].interference_number == "106125" assert mock_search.call_count == 2 # Stops when count < limit - def test_paginate_decisions_raises_on_post_body( + def test_paginate_decisions_rejects_offset_in_kwargs( self, mock_ptab_interferences_client: PTABInterferencesClient ) -> None: - """Test that paginate_decisions raises ValueError with post_body.""" - with pytest.raises(ValueError, match="does not support 'post_body'"): + """Test that paginate_decisions raises ValueError with offset in kwargs.""" + with pytest.raises(ValueError, match="Cannot specify 'offset'"): list( mock_ptab_interferences_client.paginate_decisions( - post_body={"q": "test"} + query="test", offset=10 ) ) diff --git a/tests/clients/test_ptab_trials_client.py b/tests/clients/test_ptab_trials_client.py index 99ab908..f77bb7e 100644 --- a/tests/clients/test_ptab_trials_client.py +++ b/tests/clients/test_ptab_trials_client.py @@ -859,9 +859,9 @@ def test_paginate_proceedings( assert results[2].trial_number == "IPR2023-00003" assert mock_search.call_count == 2 # Stops when count < limit - def test_paginate_proceedings_raises_on_post_body( + def test_paginate_proceedings_rejects_offset_in_kwargs( self, mock_ptab_trials_client: PTABTrialsClient ) -> None: - """Test that paginate_proceedings raises ValueError with post_body.""" - with pytest.raises(ValueError, match="does not support 'post_body'"): - list(mock_ptab_trials_client.paginate_proceedings(post_body={"q": "test"})) + """Test that paginate_proceedings raises ValueError with offset in kwargs.""" + with pytest.raises(ValueError, match="Cannot specify 'offset'"): + list(mock_ptab_trials_client.paginate_proceedings(query="test", offset=10)) From 6ab12242958bee2ed68f723b504fb11cdd1fa57b Mon Sep 17 00:00:00 2001 From: Andrew Date: Mon, 15 Dec 2025 10:59:55 -0600 Subject: [PATCH 4/4] Update tests and README examples --- README.md | 265 ++++++++++++++----------------- examples/ptab_appeals_example.py | 6 +- tests/clients/test_base.py | 209 ++++++++++++++++++------ tests/test_http_config.py | 39 +++++ 4 files changed, 321 insertions(+), 198 deletions(-) diff --git a/README.md b/README.md index f889735..3d53259 100644 --- a/README.md +++ b/README.md @@ -119,139 +119,175 @@ interferences_client = PTABInterferencesClient(config=config) ## API Usage Examples +> [!TIP] +> For comprehensive examples with detailed explanations, see the [`examples/`](examples/) directory. + ### Patent Data API ```python +from pyUSPTO import PatentDataClient + +client = PatentDataClient(api_key="your_api_key_here") + # Search for applications by inventor name -inventor_search = patent_client.search_applications(inventor_name_q="Smith") -print(f"Found {inventor_search.count} applications with 'Smith' as inventor.") -# > Found 104926 applications with 'Smith' as inventor. +response = client.search_applications(inventor_name_q="Smith", limit=2) +print(f"Found {response.count} applications with 'Smith' as inventor (showing up to 2).") + +# Get a specific application +app = client.get_application_by_number("18045436") +if app.application_meta_data: + print(f"Title: {app.application_meta_data.invention_title}") ``` +See [`examples/patent_data_example.py`](examples/patent_data_example.py) for detailed examples including downloading documents and publications. + ### Final Petition Decisions API ```python -# Search for petition decisions by date range -decisions = petition_client.search_decisions( - decision_date_from_q="2023-01-01", - limit=10 -) -print(f"Found {decisions.count} petition decisions since 2023") +from pyUSPTO import FinalPetitionDecisionsClient -# Get a specific decision by ID -decision = petition_client.get_decision_by_id("decision_id_here") -print(f"Decision Type: {decision.decision_type_code}") -print(f"Application: {decision.application_number_text}") -``` +client = FinalPetitionDecisionsClient(api_key="your_api_key_here") -### PTAB (Patent Trial and Appeal Board) APIs +# Search for petition decisions +response = client.search_decisions( + decision_date_from_q="2023-01-01", decision_date_to_q="2023-12-31", limit=5 +) +print(f"Found {response.count} decisions from 2023.") + +# Get a specific decision by ID from search results +response = client.search_decisions(limit=1) +if response.count > 0: + decision_id = response.petition_decision_data_bag[0].petition_decision_record_identifier + decision = client.get_decision_by_id(decision_id) + print(f"Decision Type: {decision.decision_type_code}") +``` -The package provides three clients for accessing PTAB data: +See [`examples/petition_decisions_example.py`](examples/petition_decisions_example.py) for detailed examples including downloading decision documents. -#### PTAB Trials API +### PTAB Trials API ```python from pyUSPTO import PTABTrialsClient -# Initialize client -trials_client = PTABTrialsClient(api_key="your_api_key_here") +client = PTABTrialsClient(api_key="your_api_key_here") -# Search for IPR trial proceedings -proceedings = trials_client.search_proceedings( +# Search for IPR proceedings +response = client.search_proceedings( trial_type_code_q="IPR", - trial_status_category_q="Instituted", petition_filing_date_from_q="2023-01-01", - limit=10 -) -print(f"Found {proceedings.count} instituted IPR proceedings") - -# Search for trial documents with new convenience parameters -documents = trials_client.search_documents( - trial_number_q="IPR2023-00001", - petitioner_party_name_q="Acme Corp", - patent_owner_name_q="XYZ Inc", - limit=5 + petition_filing_date_to_q="2023-12-31", + limit=5, ) +print(f"Found {response.count} IPR proceedings filed in 2023") -# Search for trial decisions -decisions = trials_client.search_decisions( +# Paginate through results +for proceeding in client.paginate_proceedings( trial_type_code_q="IPR", - decision_type_category_q="Final Written Decision", - patent_number_q="US1234567", - decision_date_from_q="2023-01-01" -) - -# Paginate through proceedings -for proceeding in trials_client.paginate_proceedings(trial_type_code_q="IPR", limit=25): - print(f"Trial: {proceeding.trial_number}") + petition_filing_date_from_q="2024-01-01", + limit=5, +): + print(f"{proceeding.trial_number}") ``` -#### PTAB Appeals API +See [`examples/ptab_trials_example.py`](examples/ptab_trials_example.py) for detailed examples including searching documents and decisions. + +### PTAB Appeals API ```python from pyUSPTO import PTABAppealsClient -# Initialize client -appeals_client = PTABAppealsClient(api_key="your_api_key_here") +client = PTABAppealsClient(api_key="your_api_key_here") -# Search for appeal decisions by technology center -decisions = appeals_client.search_decisions( +# Search for appeal decisions +response = client.search_decisions( technology_center_number_q="3600", - decision_type_category_q="Affirmed", decision_date_from_q="2023-01-01", - limit=10 + decision_date_to_q="2023-12-31", + limit=5, ) -print(f"Found {decisions.count} affirmed decisions from TC 3600") - -# Search by application number -decisions = appeals_client.search_decisions( - application_number_text_q="15/123456", - limit=5 -) - -# Paginate through decisions -for decision in appeals_client.paginate_decisions( - technology_center_number_q="2100", - limit=25 -): - print(f"Appeal: {decision.appeal_number}") +print(f"Found {response.count} appeal decisions from TC 3600 in 2023") ``` -#### PTAB Interferences API +See [`examples/ptab_appeals_example.py`](examples/ptab_appeals_example.py) for detailed examples including searching by decision type and application number. + +### PTAB Interferences API ```python from pyUSPTO import PTABInterferencesClient -# Initialize client -interferences_client = PTABInterferencesClient(api_key="your_api_key_here") - -# Search for interference decisions by outcome -decisions = interferences_client.search_decisions( - interference_outcome_category_q="Priority to Senior Party", - decision_date_from_q="2022-01-01", - limit=10 -) -print(f"Found {decisions.count} decisions awarding priority to senior party") +client = PTABInterferencesClient(api_key="your_api_key_here") -# Search by party name -decisions = interferences_client.search_decisions( - senior_party_name_q="Example Corp", - junior_party_name_q="Test Inc", - limit=5 +# Search for interference decisions +response = client.search_decisions( + decision_date_from_q="2023-01-01", + limit=5, ) - -# Paginate through decisions -for decision in interferences_client.paginate_decisions( - decision_type_category_q="Final Decision", - limit=25 -): - print(f"Interference: {decision.interference_number}") +print(f"Found {response.count} interference decisions since 2023") ``` +See [`examples/ptab_interferences_example.py`](examples/ptab_interferences_example.py) for detailed examples including searching by party name and outcome. + ## Documentation Full documentation may be found on [Read the Docs](https://pyuspto.readthedocs.io/). +## Data Models + +The library uses Python dataclasses to represent API responses. All data models include type annotations for attributes and methods, making them fully compatible with static type checkers. + +#### Bulk Data API + +- `BulkDataResponse`: Top-level response from the API +- `BulkDataProduct`: Information about a specific product +- `ProductFileBag`: Container for file data elements +- `FileData`: Information about an individual file + +#### Patent Data API + +- `PatentDataResponse`: Top-level response from the API +- `PatentFileWrapper`: Information about a patent application +- `ApplicationMetaData`: Metadata about a patent application +- `Address`: Represents an address in the patent data +- `Person`, `Applicant`, `Inventor`, `Attorney`: Person-related data classes +- `Assignment`, `Assignor`, `Assignee`: Assignment-related data classes +- `Continuity`, `ParentContinuity`, `ChildContinuity`: Continuity-related data classes +- `PatentTermAdjustmentData`: Patent term adjustment information +- And many more specialized classes for different aspects of patent data + +#### Final Petition Decisions API + +- `PetitionDecisionResponse`: Top-level response from the API +- `PetitionDecision`: Complete information about a petition decision +- `PetitionDecisionDocument`: Document associated with a petition decision +- `DocumentDownloadOption`: Download options for petition documents +- `DecisionTypeCode`: Enum for petition decision types +- `DocumentDirectionCategory`: Enum for document direction categories + +#### PTAB Trials API + +- `PTABTrialProceedingResponse`: Top-level response from the API +- `PTABTrialProceeding`: Information about a PTAB trial proceeding (IPR, PGR, CBM, DER) +- `PTABTrialDocument`: Document associated with a trial proceeding +- `PTABTrialDecision`: Decision information for a trial proceeding +- `RegularPetitionerData`, `RespondentData`, `DerivationPetitionerData`: Party data for different trial types +- `PTABTrialMetaData`: Trial metadata and status information + +#### PTAB Appeals API + +- `PTABAppealResponse`: Top-level response from the API +- `PTABAppealDecision`: Ex parte appeal decision information +- `AppellantData`: Appellant information and application details +- `PTABAppealMetaData`: Appeal metadata and filing information +- `PTABAppealDocumentData`: Document and decision details + +#### PTAB Interferences API + +- `PTABInterferenceResponse`: Top-level response from the API +- `PTABInterferenceDecision`: Interference proceeding decision information +- `SeniorPartyData`, `JuniorPartyData`, `AdditionalPartyData`: Party data classes +- `PTABInterferenceMetaData`: Interference metadata and status information +- `PTABInterferenceDocumentData`: Document and outcome details + ## Advanced Topics ### Advanced HTTP Configuration @@ -359,63 +395,6 @@ warnings.filterwarnings('always', category=USPTODataWarning) The library's permissive parsing philosophy returns `None` for fields that cannot be parsed, allowing you to retrieve partial data even when some fields have issues. Warnings inform you when this happens without stopping execution. -## Data Models - -The library uses Python dataclasses to represent API responses. All data models include type annotations for attributes and methods, making them fully compatible with static type checkers. - -#### Bulk Data API - -- `BulkDataResponse`: Top-level response from the API -- `BulkDataProduct`: Information about a specific product -- `ProductFileBag`: Container for file data elements -- `FileData`: Information about an individual file - -#### Patent Data API - -- `PatentDataResponse`: Top-level response from the API -- `PatentFileWrapper`: Information about a patent application -- `ApplicationMetaData`: Metadata about a patent application -- `Address`: Represents an address in the patent data -- `Person`, `Applicant`, `Inventor`, `Attorney`: Person-related data classes -- `Assignment`, `Assignor`, `Assignee`: Assignment-related data classes -- `Continuity`, `ParentContinuity`, `ChildContinuity`: Continuity-related data classes -- `PatentTermAdjustmentData`: Patent term adjustment information -- And many more specialized classes for different aspects of patent data - -#### Final Petition Decisions API - -- `PetitionDecisionResponse`: Top-level response from the API -- `PetitionDecision`: Complete information about a petition decision -- `PetitionDecisionDocument`: Document associated with a petition decision -- `DocumentDownloadOption`: Download options for petition documents -- `DecisionTypeCode`: Enum for petition decision types -- `DocumentDirectionCategory`: Enum for document direction categories - -#### PTAB Trials API - -- `PTABTrialProceedingResponse`: Top-level response from the API -- `PTABTrialProceeding`: Information about a PTAB trial proceeding (IPR, PGR, CBM, DER) -- `PTABTrialDocument`: Document associated with a trial proceeding -- `PTABTrialDecision`: Decision information for a trial proceeding -- `RegularPetitionerData`, `RespondentData`, `DerivationPetitionerData`: Party data for different trial types -- `PTABTrialMetaData`: Trial metadata and status information - -#### PTAB Appeals API - -- `PTABAppealResponse`: Top-level response from the API -- `PTABAppealDecision`: Ex parte appeal decision information -- `AppellantData`: Appellant information and application details -- `PTABAppealMetaData`: Appeal metadata and filing information -- `PTABAppealDocumentData`: Document and decision details - -#### PTAB Interferences API - -- `PTABInterferenceResponse`: Top-level response from the API -- `PTABInterferenceDecision`: Interference proceeding decision information -- `SeniorPartyData`, `JuniorPartyData`, `AdditionalPartyData`: Party data classes -- `PTABInterferenceMetaData`: Interference metadata and status information -- `PTABInterferenceDocumentData`: Document and outcome details - ## License This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details. diff --git a/examples/ptab_appeals_example.py b/examples/ptab_appeals_example.py index 5717d54..24c098b 100644 --- a/examples/ptab_appeals_example.py +++ b/examples/ptab_appeals_example.py @@ -13,7 +13,7 @@ # --- Initialization --- # Initialize the client with direct API key -print("Method 1: Initialize with direct API key") +print("Initialize with direct API key") api_key = os.environ.get("USPTO_API_KEY", "YOUR_API_KEY_HERE") if api_key == "YOUR_API_KEY_HERE": raise ValueError( @@ -182,11 +182,11 @@ decision_type_category_q="Decision", decision_date_from_q="2023-01-01", decision_date_to_q="2023-12-31", - sort="decisionDate desc", + sort="decisionData.decisionIssueDate desc", limit=3, ) - print(f"\nFound {response.count} 'Decision's from TC 2100 (Electronics) in 2023") + print(f"\nFound {response.count} Decisions from TC 2100 (Electronics) in 2023") print(f"Displaying first {len(response.patent_appeal_data_bag)} results:") for decision in response.patent_appeal_data_bag: diff --git a/tests/clients/test_base.py b/tests/clients/test_base.py index 69eaa86..ffecda7 100644 --- a/tests/clients/test_base.py +++ b/tests/clients/test_base.py @@ -402,6 +402,40 @@ def test_make_request_http_errors(self, mock_session: MagicMock) -> None: assert "This is an error less than 500 chars." in str(excinfo.value) assert excinfo.value.request_identifier is None + def test_make_request_post_error_includes_body( + self, mock_session: MagicMock + ) -> None: + """Test that POST request errors include the request body in the error message.""" + # Setup + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + client.session = mock_session + + # Mock a 400 Bad Request error + mock_response = MagicMock() + mock_response.status_code = 400 + mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError( + response=mock_response + ) + mock_response.json.return_value = { + "errorDetails": "Invalid request", + "requestIdentifier": "req-400", + } + mock_response.text = '{"errorDetails": "Invalid request"}' + + mock_session.post.return_value = mock_response + + # Make a POST request with a body + post_body = {"q": "test query", "pagination": {"limit": 100}} + + with pytest.raises(USPTOApiBadRequestError) as excinfo: + client._make_request(method="POST", endpoint="test", json_data=post_body) + + # Verify the error message includes the POST body + error_message = str(excinfo.value) + assert "Request body sent:" in error_message + assert '"q": "test query"' in error_message + assert '"pagination"' in error_message + def test_make_request_connection_error(self, mock_session: MagicMock) -> None: """Test _make_request method with connection error.""" # Setup @@ -566,43 +600,44 @@ def test_paginate_results_with_nested_pagination( self, mock_session: MagicMock ) -> None: """Test paginate_results handles nested pagination structure correctly.""" + # Setup client + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + client.session = mock_session + # Create mock responses + # response.count is the TOTAL count across all pages + # Break condition: response.count < limit + offset first_response = MagicMock() - first_response.count = 2 + first_response.count = 3 # Total: 3 items across all pages first_response.items = ["item1", "item2"] second_response = MagicMock() - second_response.count = 1 + second_response.count = 3 # Total still 3, triggers break: 3 < 2 + 2 is False, but we yield then break second_response.items = ["item3"] - third_response = MagicMock() - third_response.count = 0 - third_response.items = [] - # Track what post_body was actually sent received_bodies: list[dict[str, Any]] = [] - class TestClient(BaseUSPTOClient[Any]): - def test_method( - self, post_body: dict[str, Any] | None = None, **kwargs: Any - ) -> Any: - # Record the body we received - if post_body: - received_bodies.append(post_body.copy()) - - # Return different responses based on offset in nested pagination - if post_body and "pagination" in post_body: - offset = post_body["pagination"].get("offset", 0) - if offset == 0: - return first_response - elif offset == 2: - return second_response - else: - return third_response - return third_response + def mock_test_method( + post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Any: + """Mock method that tracks POST body and returns appropriate response.""" + # Record the body we received + if post_body: + received_bodies.append(post_body.copy()) + + # Return different responses based on offset in nested pagination + if post_body and "pagination" in post_body: + offset = post_body["pagination"].get("offset", 0) + if offset == 0: + return first_response + elif offset == 2: + return second_response + # Should not reach here + return MagicMock(count=0, items=[]) - test_client = TestClient(base_url="https://api.test.com") - test_client.session = mock_session + # Add the mock method to the client + client.test_method = mock_test_method # type: ignore[attr-defined] # Test with nested pagination structure (like USPTO API accepts) post_body = { @@ -613,7 +648,7 @@ def test_method( } results = list( - test_client.paginate_results( + client.paginate_results( method_name="test_method", response_container_attr="items", post_body=post_body, @@ -652,35 +687,28 @@ def test_paginate_results_with_flat_pagination( self, mock_session: MagicMock ) -> None: """Test paginate_results still works with flat (top-level) pagination structure.""" + # Setup client + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + client.session = mock_session + # Create mock responses + # First page: 2 items, total count shows there's only 1 item total (less than limit) first_response = MagicMock() - first_response.count = 2 - first_response.items = ["item1", "item2"] - - second_response = MagicMock() - second_response.count = 0 - second_response.items = [] + first_response.count = 1 # Total count (triggers break since 1 < 2 + 0) + first_response.items = ["item1"] received_bodies: list[dict[str, Any]] = [] - class TestClient(BaseUSPTOClient[Any]): - def test_method( - self, post_body: dict[str, Any] | None = None, **kwargs: Any - ) -> Any: - if post_body: - received_bodies.append(post_body.copy()) - - # Return different responses based on top-level offset - if post_body: - offset = post_body.get("offset", 0) - if offset == 0: - return first_response - else: - return second_response - return second_response + def mock_test_method( + post_body: dict[str, Any] | None = None, **kwargs: Any + ) -> Any: + """Mock method that tracks POST body and returns appropriate response.""" + if post_body: + received_bodies.append(post_body.copy()) + return first_response - test_client = TestClient(base_url="https://api.test.com") - test_client.session = mock_session + # Add the mock method to the client + client.test_method = mock_test_method # type: ignore[attr-defined] # Test with flat (top-level) pagination structure post_body = { @@ -689,7 +717,7 @@ def test_method( } results = list( - test_client.paginate_results( + client.paginate_results( method_name="test_method", response_container_attr="items", post_body=post_body, @@ -697,7 +725,7 @@ def test_method( ) # Verify results - assert results == ["item1", "item2"] + assert results == ["item1"] # Verify that pagination params were added at top-level assert len(received_bodies) == 1 @@ -836,6 +864,83 @@ def test_base_client_api_key_priority(self) -> None: # Explicit api_key should take precedence assert client._api_key == "explicit_key" + def test_context_manager_enters_and_exits(self, mock_session: MagicMock) -> None: + """Test that context manager __enter__ and __exit__ work correctly.""" + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + client.session = mock_session + + # Test __enter__ returns self + with client as ctx_client: + assert ctx_client is client + + # Test __exit__ was called (which calls close) + # Since we're using mock_session, we need to verify close was called on it + mock_session.close.assert_called_once() + + def test_close_when_session_is_owned(self, mock_session: MagicMock) -> None: + """Test close() closes session when client owns it.""" + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + # Client creates its own session, so it owns it + assert client._owns_session is True + + # Replace with mock for testing + client.session = mock_session + + # Close should close the session + client.close() + mock_session.close.assert_called_once() + + def test_close_when_session_is_shared(self, mock_session: MagicMock) -> None: + """Test close() does NOT close session when it's shared via config.""" + from pyUSPTO.config import USPTOConfig + + # Create config with existing shared session + config = USPTOConfig(api_key="test") + config._shared_session = mock_session + + # Create client - it should reuse the shared session and not own it + client: BaseUSPTOClient[Any] = BaseUSPTOClient( + base_url="https://api.test.com", config=config + ) + assert client._owns_session is False + + # Close should NOT close the shared session + client.close() + mock_session.close.assert_not_called() + + def test_close_backward_compatibility(self, mock_session: MagicMock) -> None: + """Test close() works when _owns_session attribute doesn't exist (backward compat).""" + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + client.session = mock_session + + # Simulate old client without _owns_session attribute + delattr(client, "_owns_session") + + # Close should still close the session for backward compatibility + client.close() + mock_session.close.assert_called_once() + + def test_paginate_results_rejects_offset_in_flat_post_body( + self, mock_session: MagicMock + ) -> None: + """Test that offset is rejected when provided in flat POST body.""" + client: BaseUSPTOClient[Any] = BaseUSPTOClient(base_url="https://api.test.com") + client.session = mock_session + + # Flat structure with user-provided offset - should raise + post_body = {"q": "test", "offset": 10, "limit": 50} + + with pytest.raises( + ValueError, match="Cannot specify 'offset' in post_body" + ): + list( + client.paginate_results( + method_name="test_method", + response_container_attr="items", + post_body=post_body, + ) + ) + class TestContentDispositionParsing: """Tests for Content-Disposition header parsing.""" diff --git a/tests/test_http_config.py b/tests/test_http_config.py index 8cf8177..e571871 100644 --- a/tests/test_http_config.py +++ b/tests/test_http_config.py @@ -110,3 +110,42 @@ def test_retry_status_codes_default(self): assert 502 in config.retry_status_codes # Bad gateway assert 503 in config.retry_status_codes # Service unavailable assert 504 in config.retry_status_codes # Gateway timeout + + def test_download_chunk_size_default(self): + """Test download_chunk_size has correct default""" + config = HTTPConfig() + assert config.download_chunk_size == 8192 + + def test_download_chunk_size_custom(self): + """Test download_chunk_size can be customized""" + config = HTTPConfig(download_chunk_size=262144) # 256 KB + assert config.download_chunk_size == 262144 + + def test_download_chunk_size_validation_negative(self): + """Test download_chunk_size validation rejects negative values""" + import pytest + + with pytest.raises(ValueError, match="download_chunk_size must be positive"): + HTTPConfig(download_chunk_size=-1) + + def test_download_chunk_size_validation_zero(self): + """Test download_chunk_size validation rejects zero""" + import pytest + + with pytest.raises(ValueError, match="download_chunk_size must be positive"): + HTTPConfig(download_chunk_size=0) + + def test_download_chunk_size_validation_large_warning(self): + """Test download_chunk_size warns for very large values""" + import pytest + + # Should warn for chunk size > 10 MB + with pytest.warns(UserWarning, match="very large and may cause memory issues"): + HTTPConfig(download_chunk_size=10485761) # 10 MB + 1 byte + + def test_download_chunk_size_from_env(self, monkeypatch): + """Test download_chunk_size can be set from environment variable""" + monkeypatch.setenv("USPTO_DOWNLOAD_CHUNK_SIZE", "131072") # 128 KB + + config = HTTPConfig.from_env() + assert config.download_chunk_size == 131072