diff --git a/CHANGELOG.md b/CHANGELOG.md index e388bfd..bdda93c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,39 @@ All notable changes to the pyUSPTO package will be documented in this file. 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.6.0] +## [Unreleased] + +### Changed + +- `sanitize_application_number` now returns series-code input as 8 digits (`18/511,739` → `18511739`). Previously it kept the slash, so `get_application_by_number` and the other per-application methods requested `.../applications/18/511739`, which the API rejects with HTTP 403. This fixes every method that takes an application number, including `get_IFW_metadata(application_number=...)` and `get_IFW(application_number=...)`. +- A 403 response with the body `{"message":"Missing Authentication Token"}` now adds to its exception message that USPTO also returns this body when the requested URL path does not exist. It is still raised as `USPTOApiAuthError`. +- `error_handling_example.py` notes that a "Missing Authentication Token" error can also mean a bad URL path. +- Every client builds its convenience filters with one shared module, `pyUSPTO.clients.query`. `PTABTrialsClient` and `PTABInterferencesClient` no longer quote a single-word text value. Live, `trialStatusCategory:Instituted` and `trialStatusCategory:"Instituted"` both return 220 proceedings. `PTABInterferencesClient` also wraps each party-name filter's `OR` group in parentheses, which returned the same records as before in live checks. + +### Fixed + +- `application_number_q` (used by `search_applications` and `get_search_results`) now goes through `sanitize_application_number`, so `14/412875`, `14/412,875`, and `14412875` all match, and `PCT/US2025/047756` is sent as `PCTUS2025047756`. Previously the value was sent verbatim and `14/412875` matched nothing. Values containing `*`, and values the sanitizer rejects (such as range or `OR` expressions), are not sanitized. +- `application_number_q` values that are not a single application number (wildcards, `OR` and range expressions) are now sent inside parentheses, so the field applies to every term. Previously it applied only to the first term: `14412875 OR 10000000` searched `10000000` as free text and returned 3 applications instead of 1. +- The application-number filters on the other clients get the same cleanup as `application_number_q`, including the parentheses above: `PetitionDecisionsClient` (`application_number_q`), `PTABAppealsClient` (`application_number_text_q`), `PTABInterferencesClient` (`senior_party_application_number_q`, `junior_party_application_number_q`), `PTABTrialsClient.search_decisions` (`application_number_q`), and `patent_application_number_q` on `OAActionsClient`, `OACitationsClient`, `OARejectionsClient`, and `EnrichedCitationsClient`. Previously `17/765301` matched nothing in any of them, and `17/765,301` matched nothing in the office action and enriched citation APIs. +- Text filters (names, titles, categories, classifications) are quoted when the value contains whitespace or one of `+ - & | ! ( ) { } [ ] ^ " ~ : \ /`, and double quotes and backslashes inside the quotes are escaped. A single plain word, including one with a `*` or `?` wildcard, is sent unquoted. A wildcard value that also contains one of those characters (`Garcia-Lopez*`) is quoted, so its wildcard is searched as plain text. Live, quoted `"Garcia-Lopez*"` returns the same 119 applications as `"Garcia-Lopez"`, and the unquoted form, sent before this change, returned HTTP 404. Before, the filters that quoted at all did so only for a space (or, for `classification_q`, a slash), so `assignee_name_q="Smith-Kline"` was sent unquoted, and an inner quote broke the phrase: `inventor_name_q='Smith "John'` matched nothing and now matches. Applies to the convenience filters of every search client. +- Code filters (such as `patent_number_q`) send a value containing spaces inside parentheses, so the field applies to every term. `patent_number_q="10000000 OR 10000001"` returned 3 applications and now returns 2. Applies to the convenience filters of every search client. +- `PTABAppealsClient` `appellant_name_q`, `requestor_name_q`, and `decision_type_category_q` were never quoted and are now quoted under the rule above. A multi-word value searched every word after the first as free text: `appellant_name_q="Block, Inc."` returned 4303 decisions from 2025 instead of 1. +- The office action and enriched citation category filters are code filters under the rule above: `OAActionsClient` `access_level_category_q` and `application_type_category_q`, `action_type_category_q` on `OACitationsClient` and `OARejectionsClient`, and `EnrichedCitationsClient` `office_action_category_q`. Their values are single-word codes (`PUBLIC`, `REGULAR`, `rejected`, `CTNF`), so `office_action_category_q="CTNF OR CTFR"` is sent as `officeActionCategory:(CTNF OR CTFR)`, and the `OR` now applies to the field. +- `customer_number_q=[]` and `status_code_q=[]` raise `ValueError` instead of sending the malformed clause `field:()`. Dropping the filter instead would have run the search unfiltered. +- `OAActionsClient` `group_art_unit_number_q=""` adds no filter. It sent `groupArtUnitNumber:`. +- `PetitionDecisionsClient.get_decision_by_id`, `BulkDataClient.get_product_by_id`, and `BulkDataClient.download_file` (when the file has no download URI) now percent-encode the ID or file name in the URL path. Previously a `?` or `/` in the value changed the request; `get_decision_by_id("abc?x")` requested decision `abc`. A non-string ID, such as a `uuid.UUID`, still works as it did before. +- `PTABTrialsClient.search_decisions(document_type_description_q=...)` escapes double quotes and backslashes in the value. A `"` in the value closed the quoted search early and broke the query. +- `BulkDataClient.download_file(file_name=...)` uses `file_name` only as the local save name. When the file had no download URI, it also requested `file_name` from the server instead of the real file, so `download_file(file_data, file_name="custom.zip")` asked for `.../custom.zip`. + +## [0.6.1] - 2026-10-27 + +### Changed + +- `requirements.txt` pins `certifi` 2026.10.23 (was 2026.7.22). +- The lint job (ruff, mypy) and the docs workflow run on Python 3.15 (was 3.14). +- Development dependency updates: certifi 2026.10.23, coverage 7.17.0, pluggy 1.7.0, pytest 9.2.0, pytest-mock 3.17.0, ruff 0.17.3, sphinx 9.2.0, sphinx-autodoc-typehints 3.17.4, sphinxcontrib-jsmath 2.0.0. + +## [0.6.0] - 2026-10-27 ### Added diff --git a/examples/error_handling_example.py b/examples/error_handling_example.py index 74ec771..31dd7b0 100644 --- a/examples/error_handling_example.py +++ b/examples/error_handling_example.py @@ -32,6 +32,7 @@ except USPTOApiAuthError as e: print(f"Authentication failed: {e}") print("Check your API key and try again.") + print("A 'Missing Authentication Token' message can also mean a bad URL path.") print("-" * 40) print("Example 2: Not found errors") @@ -95,7 +96,7 @@ except USPTOApiRateLimitError: print("Rate limited - wait and retry") except USPTOApiAuthError: - print("Authentication failed - check API key") + print("Authentication failed - check API key or URL path") except USPTOTimeout: print("Request timed out - try again") except USPTOConnectionError: diff --git a/src/pyUSPTO/clients/base.py b/src/pyUSPTO/clients/base.py index 387051b..fabb29a 100644 --- a/src/pyUSPTO/clients/base.py +++ b/src/pyUSPTO/clients/base.py @@ -14,6 +14,7 @@ TypeVar, runtime_checkable, ) +from urllib.parse import quote import requests @@ -196,6 +197,23 @@ def _build_url( base = custom_base_url if custom_base_url else self.base_url return f"{base}/{endpoint.lstrip('/')}" + @staticmethod + def _quote_path_value(value: object) -> str: + """Percent-encode a caller value for use as one URL path segment. + + Every reserved character is encoded, including ``/``, ``?``, and + ``#``, so the value cannot add path segments, a query string, or a + fragment to the request. A non-string value, such as a ``uuid.UUID``, + is formatted with ``str()`` first. + + Args: + value: The raw value, such as a record ID or file name. + + Returns: + The encoded value. + """ + return quote(str(value), safe="") + def _execute_request( self, method: str, diff --git a/src/pyUSPTO/clients/bulk_data.py b/src/pyUSPTO/clients/bulk_data.py index 0148ac3..49f5741 100644 --- a/src/pyUSPTO/clients/bulk_data.py +++ b/src/pyUSPTO/clients/bulk_data.py @@ -92,7 +92,9 @@ def get_product_by_id( ... latest=True ... ) """ - endpoint = self.ENDPOINTS["product_by_id"].format(product_id=product_id) + endpoint = self.ENDPOINTS["product_by_id"].format( + product_id=self._quote_path_value(product_id) + ) params = {} if file_data_from_date: @@ -147,7 +149,9 @@ def download_file( Args: file_data: FileData object containing download info and product_identifier. destination: Directory to save/extract to. Defaults to current directory. - file_name: Override filename. Defaults to file_data.file_name. + file_name: Name to save the file under locally. Defaults to + file_data.file_name. The download always requests + file_data.file_name from the server. overwrite: Whether to overwrite existing files. Defaults to False. extract: Whether to auto-extract archives. Defaults to False. @@ -171,8 +175,8 @@ def download_file( download_url = file_data.file_download_uri else: endpoint = self.ENDPOINTS["download_file"].format( - productIdentifier=file_data.product_identifier, - fileName=default_file_name, + productIdentifier=self._quote_path_value(file_data.product_identifier), + fileName=self._quote_path_value(file_data.file_name), ) download_url = f"{self.base_url}/{endpoint}" diff --git a/src/pyUSPTO/clients/enriched_citations.py b/src/pyUSPTO/clients/enriched_citations.py index fce7ec7..17dad06 100644 --- a/src/pyUSPTO/clients/enriched_citations.py +++ b/src/pyUSPTO/clients/enriched_citations.py @@ -9,6 +9,14 @@ from typing import Any from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + application_number, + boolean, + date_range, + phrase, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.enriched_citations import ( EnrichedCitation, @@ -84,6 +92,12 @@ def search_citations( You can provide either a direct post_body, a query string, or use convenience parameters that will be automatically combined into a query. + ``cited_document_identifier_q`` is quoted when it contains spaces or + special characters, with inner quotes escaped. The code filters + (office action and citation category codes, tech centers, art units) + wrap a value that contains spaces, such as ``"CTNF OR CTFR"``, in + parentheses, so the field applies to every term. + Args: query: Direct query string in USPTO search syntax. sort: Sort order for results. @@ -92,6 +106,8 @@ def search_citations( post_body: Optional POST body for complex queries. When provided, all other parameters are ignored. patent_application_number_q: Filter by patent application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. cited_document_identifier_q: Filter by cited document identifier. office_action_category_q: Filter by office action category (e.g., "CTNF"). citation_category_code_q: Filter by citation category code (e.g., "X", "Y"). @@ -141,41 +157,22 @@ def search_citations( # Build query from convenience parameters final_q = query if final_q is None: - q_parts = [] - if patent_application_number_q: - q_parts.append(f"patentApplicationNumber:{patent_application_number_q}") - if cited_document_identifier_q: - v = ( - f'"{cited_document_identifier_q}"' - if " " in cited_document_identifier_q - else cited_document_identifier_q - ) - q_parts.append(f"citedDocumentIdentifier:{v}") - if office_action_category_q: - q_parts.append(f"officeActionCategory:{office_action_category_q}") - if citation_category_code_q: - q_parts.append(f"citationCategoryCode:{citation_category_code_q}") - if tech_center_q: - q_parts.append(f"techCenter:{tech_center_q}") - if group_art_unit_number_q: - q_parts.append(f"groupArtUnitNumber:{group_art_unit_number_q}") - if examiner_cited_q is not None: - q_parts.append( - f"examinerCitedReferenceIndicator:{str(examiner_cited_q).lower()}" - ) - - # Handle office action date range - if office_action_date_from_q and office_action_date_to_q: - q_parts.append( - f"officeActionDate:[{office_action_date_from_q} TO {office_action_date_to_q}]" - ) - elif office_action_date_from_q: - q_parts.append(f"officeActionDate:>={office_action_date_from_q}") - elif office_action_date_to_q: - q_parts.append(f"officeActionDate:<={office_action_date_to_q}") - - if q_parts: - final_q = " AND ".join(q_parts) + final_q = all_of( + application_number( + "patentApplicationNumber", patent_application_number_q + ), + phrase("citedDocumentIdentifier", cited_document_identifier_q), + term("officeActionCategory", office_action_category_q), + term("citationCategoryCode", citation_category_code_q), + term("techCenter", tech_center_q), + term("groupArtUnitNumber", group_art_unit_number_q), + boolean("examinerCitedReferenceIndicator", examiner_cited_q), + date_range( + "officeActionDate", + office_action_date_from_q, + office_action_date_to_q, + ), + ) if final_q is not None: body["criteria"] = final_q diff --git a/src/pyUSPTO/clients/oa_actions.py b/src/pyUSPTO/clients/oa_actions.py index 5d39265..b9298e5 100644 --- a/src/pyUSPTO/clients/oa_actions.py +++ b/src/pyUSPTO/clients/oa_actions.py @@ -10,6 +10,12 @@ from typing import Any from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + application_number, + date_range, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.oa_actions import ( OAActionsFieldsResponse, @@ -77,6 +83,11 @@ def search( office action documents. You can provide either a direct post_body, a criteria string, or use convenience parameters. + Every convenience filter is a code filter (document codes, categories, + tech centers, art units). A value that contains spaces, such as + ``"rejected OR objected"``, is wrapped in parentheses, so the field + applies to every term. + Args: criteria: Direct Solr query string (e.g., ``"patentApplicationNumber:14485382"``). sort: Sort order for results (e.g., ``"submissionDate desc"``). @@ -85,6 +96,8 @@ def search( post_body: Optional POST body dict for complex queries. When provided, all other parameters are ignored. patent_application_number_q: Filter by patent application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. legacy_document_code_identifier_q: Filter by document code (e.g., ``"CTNF"``, ``"NOA"``). group_art_unit_number_q: Filter by group art unit number. tech_center_q: Filter by technology center code. @@ -132,33 +145,19 @@ def search( # Build criteria from convenience parameters final_criteria = criteria if final_criteria is None: - q_parts = [] - if patent_application_number_q: - q_parts.append(f"patentApplicationNumber:{patent_application_number_q}") - if legacy_document_code_identifier_q: - q_parts.append( - f"legacyDocumentCodeIdentifier:{legacy_document_code_identifier_q}" - ) - if group_art_unit_number_q is not None: - q_parts.append(f"groupArtUnitNumber:{group_art_unit_number_q}") - if tech_center_q: - q_parts.append(f"techCenter:{tech_center_q}") - if access_level_category_q: - q_parts.append(f"accessLevelCategory:{access_level_category_q}") - if application_type_category_q: - q_parts.append(f"applicationTypeCategory:{application_type_category_q}") - - if submission_date_from_q and submission_date_to_q: - q_parts.append( - f"submissionDate:[{submission_date_from_q} TO {submission_date_to_q}]" - ) - elif submission_date_from_q: - q_parts.append(f"submissionDate:>={submission_date_from_q}") - elif submission_date_to_q: - q_parts.append(f"submissionDate:<={submission_date_to_q}") - - if q_parts: - final_criteria = " AND ".join(q_parts) + final_criteria = all_of( + application_number( + "patentApplicationNumber", patent_application_number_q + ), + term("legacyDocumentCodeIdentifier", legacy_document_code_identifier_q), + term("groupArtUnitNumber", group_art_unit_number_q), + term("techCenter", tech_center_q), + term("accessLevelCategory", access_level_category_q), + term("applicationTypeCategory", application_type_category_q), + date_range( + "submissionDate", submission_date_from_q, submission_date_to_q + ), + ) if final_criteria is not None: body["criteria"] = final_criteria diff --git a/src/pyUSPTO/clients/oa_citations.py b/src/pyUSPTO/clients/oa_citations.py index 66a31d2..26e9b84 100644 --- a/src/pyUSPTO/clients/oa_citations.py +++ b/src/pyUSPTO/clients/oa_citations.py @@ -10,6 +10,14 @@ from typing import Any from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + application_number, + boolean, + contains, + date_range, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.oa_citations import ( OACitationRecord, @@ -79,6 +87,11 @@ def search( Office Action citation records. You can provide either a direct post_body, a criteria string, or use convenience parameters. + Every convenience filter is a code filter (document codes, categories, + tech centers, art units). A value that contains spaces, such as + ``"rejected OR objected"``, is wrapped in parentheses, so the field + applies to every term. + Args: criteria: Direct Solr query string (e.g., ``"patentApplicationNumber:16845502"``). sort: Sort order for results (e.g., ``"createDateTime desc"``). @@ -87,6 +100,8 @@ def search( post_body: Optional POST body dict for complex queries. When provided, all other parameters are ignored. patent_application_number_q: Filter by patent application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. legal_section_code_q: Filter by legal section code (e.g., ``"101"``, ``"103"``). action_type_category_q: Filter by action type (e.g., ``"rejected"``). tech_center_q: Filter by technology center code. @@ -137,37 +152,27 @@ def search( # Build criteria from convenience parameters final_criteria = criteria if final_criteria is None: - q_parts = [] - if patent_application_number_q: - q_parts.append(f"patentApplicationNumber:{patent_application_number_q}") - if legal_section_code_q: - q_parts.append(f"legalSectionCode:*{legal_section_code_q}*") - if action_type_category_q: - q_parts.append(f"actionTypeCategory:{action_type_category_q}") - if tech_center_q: - q_parts.append(f"techCenter:{tech_center_q}") - if work_group_q: - q_parts.append(f"workGroup:{work_group_q}") - if group_art_unit_number_q: - q_parts.append(f"groupArtUnitNumber:{group_art_unit_number_q}") - if examiner_cited_reference_indicator_q is not None: - val = str(examiner_cited_reference_indicator_q).lower() - q_parts.append(f"examinerCitedReferenceIndicator:{val}") - if applicant_cited_examiner_reference_indicator_q is not None: - val = str(applicant_cited_examiner_reference_indicator_q).lower() - q_parts.append(f"applicantCitedExaminerReferenceIndicator:{val}") - - if create_date_time_from_q and create_date_time_to_q: - q_parts.append( - f"createDateTime:[{create_date_time_from_q} TO {create_date_time_to_q}]" - ) - elif create_date_time_from_q: - q_parts.append(f"createDateTime:>={create_date_time_from_q}") - elif create_date_time_to_q: - q_parts.append(f"createDateTime:<={create_date_time_to_q}") - - if q_parts: - final_criteria = " AND ".join(q_parts) + final_criteria = all_of( + application_number( + "patentApplicationNumber", patent_application_number_q + ), + contains("legalSectionCode", legal_section_code_q), + term("actionTypeCategory", action_type_category_q), + term("techCenter", tech_center_q), + term("workGroup", work_group_q), + term("groupArtUnitNumber", group_art_unit_number_q), + boolean( + "examinerCitedReferenceIndicator", + examiner_cited_reference_indicator_q, + ), + boolean( + "applicantCitedExaminerReferenceIndicator", + applicant_cited_examiner_reference_indicator_q, + ), + date_range( + "createDateTime", create_date_time_from_q, create_date_time_to_q + ), + ) if final_criteria is not None: body["criteria"] = final_criteria diff --git a/src/pyUSPTO/clients/oa_rejections.py b/src/pyUSPTO/clients/oa_rejections.py index e2cfca3..6e56355 100644 --- a/src/pyUSPTO/clients/oa_rejections.py +++ b/src/pyUSPTO/clients/oa_rejections.py @@ -10,6 +10,12 @@ from typing import Any from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + application_number, + date_range, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.oa_rejections import ( OARejectionsFieldsResponse, @@ -76,6 +82,11 @@ def search( office action rejection records. You can provide either a direct post_body, a criteria string, or use convenience parameters. + Every convenience filter is a code filter (document codes, categories, + tech centers, art units). A value that contains spaces, such as + ``"rejected OR objected"``, is wrapped in parentheses, so the field + applies to every term. + Args: criteria: Direct Solr query string (e.g., ``"patentApplicationNumber:12190351"``). sort: Sort order for results (e.g., ``"submissionDate desc"``). @@ -84,6 +95,8 @@ def search( post_body: Optional POST body dict for complex queries. When provided, all other parameters are ignored. patent_application_number_q: Filter by patent application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. legacy_document_code_identifier_q: Filter by document code (e.g., ``"CTNF"``). group_art_unit_number_q: Filter by group art unit number. legal_section_code_q: Filter by legal section code. @@ -130,31 +143,18 @@ def search( # Build criteria from convenience parameters final_criteria = criteria if final_criteria is None: - q_parts = [] - if patent_application_number_q: - q_parts.append(f"patentApplicationNumber:{patent_application_number_q}") - if legacy_document_code_identifier_q: - q_parts.append( - f"legacyDocumentCodeIdentifier:{legacy_document_code_identifier_q}" - ) - if group_art_unit_number_q: - q_parts.append(f"groupArtUnitNumber:{group_art_unit_number_q}") - if legal_section_code_q: - q_parts.append(f"legalSectionCode:{legal_section_code_q}") - if action_type_category_q: - q_parts.append(f"actionTypeCategory:{action_type_category_q}") - - if submission_date_from_q and submission_date_to_q: - q_parts.append( - f"submissionDate:[{submission_date_from_q} TO {submission_date_to_q}]" - ) - elif submission_date_from_q: - q_parts.append(f"submissionDate:>={submission_date_from_q}") - elif submission_date_to_q: - q_parts.append(f"submissionDate:<={submission_date_to_q}") - - if q_parts: - final_criteria = " AND ".join(q_parts) + final_criteria = all_of( + application_number( + "patentApplicationNumber", patent_application_number_q + ), + term("legacyDocumentCodeIdentifier", legacy_document_code_identifier_q), + term("groupArtUnitNumber", group_art_unit_number_q), + term("legalSectionCode", legal_section_code_q), + term("actionTypeCategory", action_type_category_q), + date_range( + "submissionDate", submission_date_from_q, submission_date_to_q + ), + ) if final_criteria is not None: body["criteria"] = final_criteria diff --git a/src/pyUSPTO/clients/patent_data.py b/src/pyUSPTO/clients/patent_data.py index 8774044..1db63aa 100644 --- a/src/pyUSPTO/clients/patent_data.py +++ b/src/pyUSPTO/clients/patent_data.py @@ -6,7 +6,6 @@ import dataclasses import os -import re import tempfile import warnings import zipfile @@ -16,6 +15,16 @@ import requests from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + any_of, + application_number, + date_range, + phrase, + publication_number, + sanitize_application_number, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.exceptions import FormatNotAvailableError, USPTOApiNotFoundError from pyUSPTO.models.patent_data import ( @@ -39,42 +48,6 @@ ) from pyUSPTO.warnings import USPTODataMismatchWarning - -def _or_clause(field: str, value: int | str | list[int | str]) -> str: - """Build a Lucene clause for a single value or a list of values. - - Single value → ``field:val``. List → ``field:(v1 OR v2 ...)``. - """ - if isinstance(value, list): - return f"{field}:({' OR '.join(str(v) for v in value)})" - return f"{field}:{value}" - - -_PUBLICATION_NUMBER_RE = re.compile( - r"^(?:US)?[\s/\-,]*(\d{4})[\s/\-,]*(\d{7})[\s/\-,]*(?:[A-Z]\d)?$" -) - - -def _publication_number_query(value: str) -> str: - """Build the search value for ``applicationMetaData.earliestPublicationNumber``. - - The API stores publication numbers as ``USyyyyNNNNNNNAn`` (e.g. - ``US20210138233A1``). Callers typically supply bare digits, or forms with - separators such as ``2021/0138233`` or ``US 2021/0138233 A1``. This reduces - the input to its 11 digits (year + 7-digit serial) and wraps them in - wildcards so any prefix or kind code matches. - - Values that already contain a wildcard, or that do not look like a US - publication number, are returned unchanged. - """ - if "*" in value: - return value - match = _PUBLICATION_NUMBER_RE.match(value.strip().upper()) - if match is None: - return value - return f"*{match.group(1)}{match.group(2)}*" - - FIELD_PRESETS: dict[str, list[str]] = { "minimal": [ "applicationNumberText", @@ -173,16 +146,16 @@ def sanitize_application_number(self, input_number: str) -> str: - PCT format: "PCT/US2024/012345" → "PCTUS2024012345" (standardized 15-character form per USPTO ODP Release 3.6, 2026-04-10) - This method removes common separators (commas, spaces) while preserving - the "/" in series code format. + This method removes common separators (commas, spaces) and joins + series-code input into 8 digits. Args: input_number: Raw application number input. May include commas, spaces, or other formatting. Returns: - str: Sanitized application number ("NNNNNNNN", "NN/NNNNNN", or - "PCT" + 2-char country + 4-digit year + 6-digit serial). + str: Sanitized application number ("NNNNNNNN", or "PCT" + 2-char + country + 4-digit year + 6-digit serial). Raises: ValueError: If the format is invalid. @@ -193,115 +166,13 @@ def sanitize_application_number(self, input_number: str) -> str: >>> client.sanitize_application_number("16,123,456") "16123456" >>> client.sanitize_application_number("08/123456") - "08/123456" + "08123456" >>> client.sanitize_application_number("08/123,456") - "08/123456" + "08123456" >>> client.sanitize_application_number("PCT/US2024/012345") "PCTUS2024012345" """ - if not input_number or not input_number.strip(): - raise ValueError("Application number cannot be empty") - - raw = input_number.strip() - - # Handle PCT formats. Output is the standardized 15-char form: - # "PCT" + 2-char country + 4-digit year + 6-digit zero-padded serial. - # Example: "PCT/US2024/012345" -> "PCTUS2024012345" - if raw.startswith("PCT"): - parts = raw.split("/") - if len(parts) == 1: - # No slashes — only accept input that is already in the 15-char form. - body = raw[3:] - if len(body) == 12 and body[:2].isalpha() and body[2:].isdigit(): - return raw - raise ValueError( - f"Invalid PCT application format: {input_number}. " - "Expected PCT/CCYYYY/NNNNNN or PCTCCYYYYNNNNNN." - ) - if len(parts) != 3: - raise ValueError( - f"Invalid PCT application format: {input_number}. " - "Expected PCT/CCYYYY/NNNNNN" - ) - - _, country_year, serial = parts - - # country_year can be "US2024" or "US24" - country = country_year[:2] - if len(country) != 2 or not country.isalpha(): - raise ValueError( - f"Invalid PCT country code in: {country_year}. Expected 2 letters." - ) - - year_part = country_year[2:] - if not year_part.isdigit(): - raise ValueError( - f"Invalid PCT year in: {country_year}. Must be digits." - ) - - # Normalize to 4-digit year. 2-digit input uses a sliding window - # tied to the PCT system's start year (1978): YY>=78 -> 19YY, else 20YY. - if len(year_part) == 4: - year = year_part - elif len(year_part) == 2: - yy = int(year_part) - year = f"19{year_part}" if yy >= 78 else f"20{year_part}" - else: - raise ValueError( - f"Invalid PCT year length in: {country_year}. " - "Expected CCYYYY or CCYY." - ) - - # Serial must be digits only, and fit in 6 digits. - if not serial.isdigit(): - raise ValueError(f"Invalid PCT serial: {serial}. Must be numeric.") - if len(serial) > 6: - raise ValueError( - f"Invalid PCT serial: {serial}. Must be at most 6 digits." - ) - - return f"PCT{country}{year}{serial.zfill(6)}" - - # Strip whitespace and remove commas/spaces - cleaned = raw.replace(",", "").replace(" ", "") - - # Check if this is series code format (NN/NNNNNN) - if "/" in cleaned: - parts = cleaned.split("/") - if len(parts) != 2: - raise ValueError( - f"Invalid application number format: {input_number}. " - "Expected format: NNNNNNNN or NN/NNNNNN" - ) - - series, serial = parts - if not series.isdigit() or not serial.isdigit(): - raise ValueError( - f"Invalid application number format: {input_number}. " - "Series and serial must be numeric." - ) - - if len(series) != 2 or len(serial) != 6: - raise ValueError( - f"Invalid application number format: {input_number}. " - "Expected series code format: NN/NNNNNN (2 digits / 6 digits)" - ) - - return cleaned - - # Standard 8-digit format - if not cleaned.isdigit(): - raise ValueError( - f"Invalid application number format: {input_number}. " - "Must contain only digits." - ) - - if len(cleaned) != 8: - raise ValueError( - f"Invalid application number format: {input_number}. Expected 8 digits." - ) - - return cleaned + return sanitize_application_number(input_number) def _get_wrapper_from_response( self, @@ -328,6 +199,54 @@ def _get_wrapper_from_response( ) return wrapper + def _search_query( + self, + *, + application_number_q: str | None, + patent_number_q: str | None, + inventor_name_q: str | None, + applicant_name_q: str | None, + assignee_name_q: str | None, + customer_number_q: int | str | list[int | str] | None, + status_code_q: int | str | list[int | str] | None, + filing_date_from_q: str | None, + filing_date_to_q: str | None, + grant_date_from_q: str | None, + grant_date_to_q: str | None, + status_date_from_q: str | None, + status_date_to_q: str | None, + classification_q: str | None, + earliestPublicationNumber_q: str | None = None, + pctPublicationNumber_q: str | None = None, + ) -> str | None: + """Build the ``q`` value for the application search convenience filters.""" + return all_of( + application_number("applicationNumberText", application_number_q), + term("applicationMetaData.patentNumber", patent_number_q), + phrase("applicationMetaData.inventorBag.inventorNameText", inventor_name_q), + phrase("applicationMetaData.firstApplicantName", applicant_name_q), + phrase("assignmentBag.assigneeBag.assigneeNameText", assignee_name_q), + any_of("applicationMetaData.customerNumber", customer_number_q), + any_of("applicationMetaData.applicationStatusCode", status_code_q), + phrase("applicationMetaData.cpcClassificationBag", classification_q), + publication_number( + "applicationMetaData.earliestPublicationNumber", + earliestPublicationNumber_q, + ), + term("applicationMetaData.pctPublicationNumber", pctPublicationNumber_q), + date_range( + "applicationMetaData.filingDate", filing_date_from_q, filing_date_to_q + ), + date_range( + "applicationMetaData.grantDate", grant_date_from_q, grant_date_to_q + ), + date_range( + "applicationMetaData.applicationStatusDate", + status_date_from_q, + status_date_to_q, + ), + ) + def search_applications( self, query: str | None = None, @@ -362,10 +281,24 @@ def search_applications( Can perform a GET request based on OpenAPI query parameters or a POST request if post_body is specified. ``customer_number_q`` and ``status_code_q`` accept a single value or a list; - a list builds a Lucene ``field:(a OR b)`` clause. ``status_date_from_q`` / + a list builds a Lucene ``field:(a OR b)`` clause, and an empty list raises + ``ValueError``. ``status_date_from_q`` / ``status_date_to_q`` follow the same date-range pattern as ``filing_date_from_q`` / ``filing_date_to_q``. + Text filters (names, classification) are quoted when they contain spaces + or special characters, with inner quotes escaped. Code filters (such as + ``patent_number_q``) that contain spaces are wrapped in parentheses, so + the field applies to every term. + + ``application_number_q`` is cleaned with ``sanitize_application_number``, + so ``14/412,875``, ``14/412875``, and ``14412875`` all search for + ``14412875``, and ``PCT/US2025/047756`` searches for ``PCTUS2025047756``. + A value that contains ``*``, or that is not a single application number + (such as an ``OR`` or range expression), is sent inside parentheses, so + the field applies to every term. In an ``OR`` or range expression, write + each number as 8 digits. + ``earliestPublicationNumber_q`` is reduced to its digits and searched with wildcards (``*20210138233*``), so bare digits, ``US`` prefixes, kind codes, and separators all match the stored ``USyyyyNNNNNNNAn`` @@ -386,103 +319,24 @@ def search_applications( final_q = query if final_q is None: - q_parts = [] - if application_number_q: - q_parts.append(f"applicationNumberText:{application_number_q}") - if patent_number_q: - q_parts.append( - f"applicationMetaData.patentNumber:{patent_number_q}" - ) - if inventor_name_q: - v = ( - f'"{inventor_name_q}"' - if " " in inventor_name_q - else inventor_name_q - ) - q_parts.append( - f"applicationMetaData.inventorBag.inventorNameText:{v}" - ) - if applicant_name_q: - v = ( - f'"{applicant_name_q}"' - if " " in applicant_name_q - else applicant_name_q - ) - q_parts.append(f"applicationMetaData.firstApplicantName:{v}") - if assignee_name_q: - v = ( - f'"{assignee_name_q}"' - if " " in assignee_name_q - else assignee_name_q - ) - q_parts.append(f"assignmentBag.assigneeBag.assigneeNameText:{v}") - if customer_number_q is not None: - q_parts.append( - _or_clause( - "applicationMetaData.customerNumber", customer_number_q - ) - ) - if status_code_q is not None: - q_parts.append( - _or_clause( - "applicationMetaData.applicationStatusCode", status_code_q - ) - ) - if classification_q: - v = ( - f'"{classification_q}"' - if any(c in classification_q for c in [" ", "/"]) - else classification_q - ) - q_parts.append(f"applicationMetaData.cpcClassificationBag:{v}") - if earliestPublicationNumber_q: - q_parts.append( - "applicationMetaData.earliestPublicationNumber:" - f"{_publication_number_query(earliestPublicationNumber_q)}" - ) - if pctPublicationNumber_q: - q_parts.append( - f"applicationMetaData.pctPublicationNumber:{pctPublicationNumber_q}" - ) - if filing_date_from_q and filing_date_to_q: - q_parts.append( - f"applicationMetaData.filingDate:[{filing_date_from_q} TO {filing_date_to_q}]" - ) - elif filing_date_from_q: - q_parts.append( - f"applicationMetaData.filingDate:>={filing_date_from_q}" - ) - elif filing_date_to_q: - q_parts.append( - f"applicationMetaData.filingDate:<={filing_date_to_q}" - ) - - if grant_date_from_q and grant_date_to_q: - q_parts.append( - f"applicationMetaData.grantDate:[{grant_date_from_q} TO {grant_date_to_q}]" - ) - elif grant_date_from_q: - q_parts.append( - f"applicationMetaData.grantDate:>={grant_date_from_q}" - ) - elif grant_date_to_q: - q_parts.append(f"applicationMetaData.grantDate:<={grant_date_to_q}") - - if status_date_from_q and status_date_to_q: - q_parts.append( - f"applicationMetaData.applicationStatusDate:[{status_date_from_q} TO {status_date_to_q}]" - ) - elif status_date_from_q: - q_parts.append( - f"applicationMetaData.applicationStatusDate:>={status_date_from_q}" - ) - elif status_date_to_q: - q_parts.append( - f"applicationMetaData.applicationStatusDate:<={status_date_to_q}" - ) - - if q_parts: - final_q = " AND ".join(q_parts) + final_q = self._search_query( + application_number_q=application_number_q, + patent_number_q=patent_number_q, + inventor_name_q=inventor_name_q, + applicant_name_q=applicant_name_q, + assignee_name_q=assignee_name_q, + customer_number_q=customer_number_q, + status_code_q=status_code_q, + filing_date_from_q=filing_date_from_q, + filing_date_to_q=filing_date_to_q, + grant_date_from_q=grant_date_from_q, + grant_date_to_q=grant_date_to_q, + status_date_from_q=status_date_from_q, + status_date_to_q=status_date_to_q, + classification_q=classification_q, + earliestPublicationNumber_q=earliestPublicationNumber_q, + pctPublicationNumber_q=pctPublicationNumber_q, + ) if final_q is not None: params["q"] = final_q @@ -541,6 +395,10 @@ def get_search_results( For GET, parameters align with OpenAPI for /api/v1/patent/applications/search/download. For POST, post_body should conform to PatentDownloadRequest schema. + + The convenience filters (``application_number_q`` through + ``classification_q``) are built the same way as in + ``search_applications``. """ endpoint = self.ENDPOINTS["get_search_results"] @@ -559,95 +417,22 @@ def get_search_results( final_q = query if final_q is None: - q_parts = [] - if application_number_q: - q_parts.append(f"applicationNumberText:{application_number_q}") - if patent_number_q: - q_parts.append( - f"applicationMetaData.patentNumber:{patent_number_q}" - ) - if inventor_name_q: - v = ( - f'"{inventor_name_q}"' - if " " in inventor_name_q - else inventor_name_q - ) - q_parts.append( - f"applicationMetaData.inventorBag.inventorNameText:{v}" - ) - if applicant_name_q: - v = ( - f'"{applicant_name_q}"' - if " " in applicant_name_q - else applicant_name_q - ) - q_parts.append(f"applicationMetaData.firstApplicantName:{v}") - if assignee_name_q: - v = ( - f'"{assignee_name_q}"' - if " " in assignee_name_q - else assignee_name_q - ) - q_parts.append(f"assignmentBag.assigneeBag.assigneeNameText:{v}") - if customer_number_q is not None: - q_parts.append( - _or_clause( - "applicationMetaData.customerNumber", customer_number_q - ) - ) - if status_code_q is not None: - q_parts.append( - _or_clause( - "applicationMetaData.applicationStatusCode", status_code_q - ) - ) - if classification_q: - v = ( - f'"{classification_q}"' - if any(c in classification_q for c in [" ", "/"]) - else classification_q - ) - q_parts.append(f"applicationMetaData.cpcClassificationBag:{v}") - - if filing_date_from_q and filing_date_to_q: - q_parts.append( - f"applicationMetaData.filingDate:[{filing_date_from_q} TO {filing_date_to_q}]" - ) - elif filing_date_from_q: - q_parts.append( - f"applicationMetaData.filingDate:>={filing_date_from_q}" - ) - elif filing_date_to_q: - q_parts.append( - f"applicationMetaData.filingDate:<={filing_date_to_q}" - ) - - if grant_date_from_q and grant_date_to_q: - q_parts.append( - f"applicationMetaData.grantDate:[{grant_date_from_q} TO {grant_date_to_q}]" - ) - elif grant_date_from_q: - q_parts.append( - f"applicationMetaData.grantDate:>={grant_date_from_q}" - ) - elif grant_date_to_q: - q_parts.append(f"applicationMetaData.grantDate:<={grant_date_to_q}") - - if status_date_from_q and status_date_to_q: - q_parts.append( - f"applicationMetaData.applicationStatusDate:[{status_date_from_q} TO {status_date_to_q}]" - ) - elif status_date_from_q: - q_parts.append( - f"applicationMetaData.applicationStatusDate:>={status_date_from_q}" - ) - elif status_date_to_q: - q_parts.append( - f"applicationMetaData.applicationStatusDate:<={status_date_to_q}" - ) - - if q_parts: - final_q = " AND ".join(q_parts) + final_q = self._search_query( + application_number_q=application_number_q, + patent_number_q=patent_number_q, + inventor_name_q=inventor_name_q, + applicant_name_q=applicant_name_q, + assignee_name_q=assignee_name_q, + customer_number_q=customer_number_q, + status_code_q=status_code_q, + filing_date_from_q=filing_date_from_q, + filing_date_to_q=filing_date_to_q, + grant_date_from_q=grant_date_from_q, + grant_date_to_q=grant_date_to_q, + status_date_from_q=status_date_from_q, + status_date_to_q=status_date_to_q, + classification_q=classification_q, + ) if final_q is not None: params["q"] = final_q @@ -690,8 +475,9 @@ def get_application_by_number( Args: application_number (str): The USPTO application number for the patent - application (e.g., "16123456" or "18/915,708"). The application - number will be automatically sanitized to remove commas and spaces. + application (e.g., "16123456" or "18/915,708"). The number is + cleaned with ``sanitize_application_number``, so "18/915,708" + is requested as "18915708". Returns: Optional[PatentFileWrapper]: A `PatentFileWrapper` object representing diff --git a/src/pyUSPTO/clients/petition_decisions.py b/src/pyUSPTO/clients/petition_decisions.py index ed797e3..b48cf86 100644 --- a/src/pyUSPTO/clients/petition_decisions.py +++ b/src/pyUSPTO/clients/petition_decisions.py @@ -12,6 +12,13 @@ import requests from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + application_number, + date_range, + phrase, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.petition_decisions import ( DocumentDownloadOption, @@ -142,6 +149,38 @@ def _add_fixed_filters_to_body(self, post_body: dict[str, Any]) -> dict[str, Any caller_filters = [] if existing is None else existing return {**post_body, "filters": [*caller_filters, *fixed]} + def _decisions_query( + self, + *, + application_number_q: str | None, + patent_number_q: str | None, + inventor_name_q: str | None, + applicant_name_q: str | None, + decision_date_from_q: str | None, + decision_date_to_q: str | None, + invention_title_q: str | None = None, + decided_petition_type_code_q: str | None = None, + technology_center_q: str | None = None, + final_deciding_office_name_q: str | None = None, + petition_mail_date_from_q: str | None = None, + petition_mail_date_to_q: str | None = None, + ) -> str | None: + """Build the ``q`` value for the petition decision convenience filters.""" + return all_of( + application_number("applicationNumberText", application_number_q), + term("patentNumber", patent_number_q), + phrase("inventorBag", inventor_name_q), + phrase("firstApplicantName", applicant_name_q), + phrase("inventionTitle", invention_title_q), + term("decidedPetitionTypeCode", decided_petition_type_code_q), + term("technologyCenter", technology_center_q), + phrase("finalDecidingOfficeName", final_deciding_office_name_q), + date_range("decisionDate", decision_date_from_q, decision_date_to_q), + date_range( + "petitionMailDate", petition_mail_date_from_q, petition_mail_date_to_q + ), + ) + def search_decisions( self, query: str | None = None, @@ -175,6 +214,11 @@ def search_decisions( direct query string or use convenience parameters that will be automatically combined into a query. + Text filters (names, titles, office names) are quoted when they contain + spaces or special characters, with inner quotes escaped. Code filters + (patent number, type code, technology center) that contain spaces are + wrapped in parentheses, so the field applies to every term. + Args: query: Direct query string in USPTO search syntax. sort: Sort as "field order", e.g. "decisionDate desc". A list sends one @@ -189,6 +233,8 @@ def search_decisions( range_filters: Range filter configuration string. post_body: Optional POST body for complex queries. application_number_q: Filter by application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. patent_number_q: Filter by patent number. inventor_name_q: Filter by inventor name. applicant_name_q: Filter by applicant name. @@ -238,70 +284,21 @@ def search_decisions( params: dict[str, Any] = {} final_q = query - # Build query from convenience parameters if final_q is None: - q_parts = [] - if application_number_q: - q_parts.append(f"applicationNumberText:{application_number_q}") - if patent_number_q: - q_parts.append(f"patentNumber:{patent_number_q}") - if inventor_name_q: - v = ( - f'"{inventor_name_q}"' - if " " in inventor_name_q - else inventor_name_q - ) - q_parts.append(f"inventorBag:{v}") - if applicant_name_q: - v = ( - f'"{applicant_name_q}"' - if " " in applicant_name_q - else applicant_name_q - ) - q_parts.append(f"firstApplicantName:{v}") - if invention_title_q: - v = ( - f'"{invention_title_q}"' - if " " in invention_title_q - else invention_title_q - ) - q_parts.append(f"inventionTitle:{v}") - if decided_petition_type_code_q: - q_parts.append( - f"decidedPetitionTypeCode:{decided_petition_type_code_q}" - ) - if technology_center_q: - q_parts.append(f"technologyCenter:{technology_center_q}") - if final_deciding_office_name_q: - v = ( - f'"{final_deciding_office_name_q}"' - if " " in final_deciding_office_name_q - else final_deciding_office_name_q - ) - q_parts.append(f"finalDecidingOfficeName:{v}") - - # Handle decision date range - if decision_date_from_q and decision_date_to_q: - q_parts.append( - f"decisionDate:[{decision_date_from_q} TO {decision_date_to_q}]" - ) - elif decision_date_from_q: - q_parts.append(f"decisionDate:>={decision_date_from_q}") - elif decision_date_to_q: - q_parts.append(f"decisionDate:<={decision_date_to_q}") - - # Handle petition mail date range - if petition_mail_date_from_q and petition_mail_date_to_q: - q_parts.append( - f"petitionMailDate:[{petition_mail_date_from_q} TO {petition_mail_date_to_q}]" - ) - elif petition_mail_date_from_q: - q_parts.append(f"petitionMailDate:>={petition_mail_date_from_q}") - elif petition_mail_date_to_q: - q_parts.append(f"petitionMailDate:<={petition_mail_date_to_q}") - - if q_parts: - final_q = " AND ".join(q_parts) + final_q = self._decisions_query( + application_number_q=application_number_q, + patent_number_q=patent_number_q, + inventor_name_q=inventor_name_q, + applicant_name_q=applicant_name_q, + decision_date_from_q=decision_date_from_q, + decision_date_to_q=decision_date_to_q, + invention_title_q=invention_title_q, + decided_petition_type_code_q=decided_petition_type_code_q, + technology_center_q=technology_center_q, + final_deciding_office_name_q=final_deciding_office_name_q, + petition_mail_date_from_q=petition_mail_date_from_q, + petition_mail_date_to_q=petition_mail_date_to_q, + ) # Add parameters if final_q is not None: @@ -356,7 +353,9 @@ def get_decision_by_id( ... ) """ endpoint = self.ENDPOINTS["get_decision_by_id"].format( - petitionDecisionRecordIdentifier=petition_decision_record_identifier + petitionDecisionRecordIdentifier=self._quote_path_value( + petition_decision_record_identifier + ) ) response_data = self._get_model( @@ -398,6 +397,11 @@ def download_decisions( This endpoint is designed for bulk downloads of petition decisions data. It supports JSON and CSV formats. + Text filters (names, titles, office names) are quoted when they contain + spaces or special characters, with inner quotes escaped. Code filters + (patent number, type code, technology center) that contain spaces are + wrapped in parentheses, so the field applies to every term. + Args: format: Download format, either "json" or "csv". Defaults to "json". query: Direct query string in USPTO search syntax. @@ -410,6 +414,8 @@ def download_decisions( A value containing a space only works in post_body filters. range_filters: Range filter configuration string. application_number_q: Filter by application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. patent_number_q: Filter by patent number. inventor_name_q: Filter by inventor name. applicant_name_q: Filter by applicant name. @@ -458,40 +464,15 @@ def download_decisions( params: dict[str, Any] = {"format": format} final_q = query - # Build query from convenience parameters if final_q is None: - q_parts = [] - if application_number_q: - q_parts.append(f"applicationNumberText:{application_number_q}") - if patent_number_q: - q_parts.append(f"patentNumber:{patent_number_q}") - if inventor_name_q: - v = ( - f'"{inventor_name_q}"' - if " " in inventor_name_q - else inventor_name_q - ) - q_parts.append(f"inventorBag:{v}") - if applicant_name_q: - v = ( - f'"{applicant_name_q}"' - if " " in applicant_name_q - else applicant_name_q - ) - q_parts.append(f"firstApplicantName:{v}") - - # Handle decision date range - if decision_date_from_q and decision_date_to_q: - q_parts.append( - f"decisionDate:[{decision_date_from_q} TO {decision_date_to_q}]" - ) - elif decision_date_from_q: - q_parts.append(f"decisionDate:>={decision_date_from_q}") - elif decision_date_to_q: - q_parts.append(f"decisionDate:<={decision_date_to_q}") - - if q_parts: - final_q = " AND ".join(q_parts) + final_q = self._decisions_query( + application_number_q=application_number_q, + patent_number_q=patent_number_q, + inventor_name_q=inventor_name_q, + applicant_name_q=applicant_name_q, + decision_date_from_q=decision_date_from_q, + decision_date_to_q=decision_date_to_q, + ) # Add parameters if final_q is not None: diff --git a/src/pyUSPTO/clients/ptab_appeals.py b/src/pyUSPTO/clients/ptab_appeals.py index c6f5f83..03582a4 100644 --- a/src/pyUSPTO/clients/ptab_appeals.py +++ b/src/pyUSPTO/clients/ptab_appeals.py @@ -8,6 +8,13 @@ from typing import Any from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + application_number, + date_range, + phrase, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.ptab import ( AppealDocumentData, @@ -88,6 +95,11 @@ def search_decisions( direct query string or use convenience parameters that will be automatically combined into a query. + Text filters (names, decision type) are quoted when they contain spaces + or special characters, with inner quotes escaped. Code filters (appeal + number, technology center) that contain spaces are wrapped in + parentheses, so the field applies to every term. + Args: query: Direct query string in USPTO search syntax. sort: Sort order for results. @@ -102,6 +114,8 @@ def search_decisions( post_body: Optional POST body for complex queries. appeal_number_q: Filter by appeal number. application_number_text_q: Filter by application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. appellant_name_q: Filter by appellant name. requestor_name_q: Filter by requestor name. decision_type_category_q: Filter by decision type category. @@ -150,44 +164,26 @@ def search_decisions( # Build query from convenience parameters if final_q is None: - q_parts = [] - if appeal_number_q: - q_parts.append(f"appealNumber:{appeal_number_q}") - if application_number_text_q: - q_parts.append( - f"appellantData.applicationNumberText:{application_number_text_q}" - ) - if appellant_name_q: - q_parts.append( - f"appellantData.realPartyInInterestName:{appellant_name_q}" - ) - if requestor_name_q: - q_parts.append(f"appellantData.counselName:{requestor_name_q}") - if decision_type_category_q: - q_parts.append( - f"decisionData.decisionTypeCategory:{decision_type_category_q}" - ) - if technology_center_number_q: - q_parts.append( - f"appellantData.technologyCenterNumber:{technology_center_number_q}" - ) - - # Handle decision date range - if decision_date_from_q and decision_date_to_q: - q_parts.append( - f"decisionData.decisionIssueDate:[{decision_date_from_q} TO {decision_date_to_q}]" - ) - elif decision_date_from_q: - q_parts.append( - f"decisionData.decisionIssueDate:>={decision_date_from_q}" - ) - elif decision_date_to_q: - q_parts.append( - f"decisionData.decisionIssueDate:<={decision_date_to_q}" - ) - - if q_parts: - final_q = " AND ".join(q_parts) + final_q = all_of( + term("appealNumber", appeal_number_q), + application_number( + "appellantData.applicationNumberText", application_number_text_q + ), + phrase("appellantData.realPartyInInterestName", appellant_name_q), + phrase("appellantData.counselName", requestor_name_q), + phrase( + "decisionData.decisionTypeCategory", decision_type_category_q + ), + term( + "appellantData.technologyCenterNumber", + technology_center_number_q, + ), + date_range( + "decisionData.decisionIssueDate", + decision_date_from_q, + decision_date_to_q, + ), + ) # Add parameters if final_q is not None: diff --git a/src/pyUSPTO/clients/ptab_interferences.py b/src/pyUSPTO/clients/ptab_interferences.py index 6e0ff7c..a95fc94 100644 --- a/src/pyUSPTO/clients/ptab_interferences.py +++ b/src/pyUSPTO/clients/ptab_interferences.py @@ -8,6 +8,14 @@ from typing import Any from pyUSPTO.clients.base import BaseUSPTOClient +from pyUSPTO.clients.query import ( + all_of, + application_number, + date_range, + either, + phrase, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.ptab import ( InterferenceDocumentData, @@ -90,6 +98,12 @@ def search_decisions( direct query string or use convenience parameters that will be automatically combined into a query. + Text filters (party names, categories) are quoted when they contain + spaces or special characters, with inner quotes escaped. A party-name + filter searches several name fields inside one parenthesized OR group. + The interference number filter, if it contains spaces, is wrapped in + parentheses, so the field applies to every term. + Args: query: Direct query string in USPTO search syntax. sort: Sort order for results. @@ -104,7 +118,11 @@ def search_decisions( post_body: Optional POST body for complex queries. interference_number_q: Filter by interference number. senior_party_application_number_q: Filter by senior party application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. junior_party_application_number_q: Filter by junior party application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. senior_party_name_q: Filter by senior party name. junior_party_name_q: Filter by junior party name. real_party_in_interest_q: Filter by Real Party in Interest. @@ -154,55 +172,55 @@ def search_decisions( # Build query from convenience parameters if final_q is None: - q_parts = [] - if interference_number_q: - q_parts.append(f"interferenceNumber:{interference_number_q}") - if senior_party_application_number_q: - q_parts.append( - f"seniorPartyData.applicationNumberText:{senior_party_application_number_q}" - ) - if junior_party_application_number_q: - q_parts.append( - f"juniorPartyData.applicationNumberText:{junior_party_application_number_q}" - ) - if senior_party_name_q: - q_parts.append( - f'seniorPartyData.patentOwnerName:"{senior_party_name_q}" OR seniorPartyData.inventorName:"{senior_party_name_q}" OR seniorPartyData.realPartyInInterestName:"{senior_party_name_q}"' - ) - if junior_party_name_q: - q_parts.append( - f'juniorPartyData.patentOwnerName:"{junior_party_name_q}" OR juniorPartyData.inventorName:"{junior_party_name_q}" OR juniorPartyData.realPartyInInterestName:"{junior_party_name_q}"' - ) - if real_party_in_interest_q: - q_parts.append( - f'seniorPartyData.realPartyInInterestName:"{real_party_in_interest_q}" OR juniorPartyData.realPartyInInterestName:"{real_party_in_interest_q}"' - ) - - if interference_outcome_category_q: - q_parts.append( - f'documentData.interferenceOutcomeCategory:"{interference_outcome_category_q}"' - ) - if decision_type_category_q: - q_parts.append( - f'documentData.decisionTypeCategory:"{decision_type_category_q}"' - ) - - # Handle decision date range - if decision_date_from_q and decision_date_to_q: - q_parts.append( - f"documentData.decisionIssueDate:[{decision_date_from_q} TO {decision_date_to_q}]" - ) - elif decision_date_from_q: - q_parts.append( - f"documentData.decisionIssueDate:>={decision_date_from_q}" - ) - elif decision_date_to_q: - q_parts.append( - f"documentData.decisionIssueDate:<={decision_date_to_q}" - ) - - if q_parts: - final_q = " AND ".join(q_parts) + final_q = all_of( + term("interferenceNumber", interference_number_q), + application_number( + "seniorPartyData.applicationNumberText", + senior_party_application_number_q, + ), + application_number( + "juniorPartyData.applicationNumberText", + junior_party_application_number_q, + ), + either( + phrase("seniorPartyData.patentOwnerName", senior_party_name_q), + phrase("seniorPartyData.inventorName", senior_party_name_q), + phrase( + "seniorPartyData.realPartyInInterestName", + senior_party_name_q, + ), + ), + either( + phrase("juniorPartyData.patentOwnerName", junior_party_name_q), + phrase("juniorPartyData.inventorName", junior_party_name_q), + phrase( + "juniorPartyData.realPartyInInterestName", + junior_party_name_q, + ), + ), + either( + phrase( + "seniorPartyData.realPartyInInterestName", + real_party_in_interest_q, + ), + phrase( + "juniorPartyData.realPartyInInterestName", + real_party_in_interest_q, + ), + ), + phrase( + "documentData.interferenceOutcomeCategory", + interference_outcome_category_q, + ), + phrase( + "documentData.decisionTypeCategory", decision_type_category_q + ), + date_range( + "documentData.decisionIssueDate", + decision_date_from_q, + decision_date_to_q, + ), + ) # Add parameters if final_q is not None: diff --git a/src/pyUSPTO/clients/ptab_trials.py b/src/pyUSPTO/clients/ptab_trials.py index eeed9bd..b84266a 100644 --- a/src/pyUSPTO/clients/ptab_trials.py +++ b/src/pyUSPTO/clients/ptab_trials.py @@ -9,6 +9,14 @@ from typing import Any from pyUSPTO.clients.base import BaseUSPTOClient, M +from pyUSPTO.clients.query import ( + all_of, + application_number, + contains, + date_range, + phrase, + term, +) from pyUSPTO.config import USPTOConfig from pyUSPTO.models.ptab import ( PTABTrialDocumentResponse, @@ -70,7 +78,7 @@ def _perform_search( endpoint_key: str, response_class: type[M], query: str | None, - query_parts: list[str], + built_query: str | None, post_body: dict[str, Any] | None, sort: str | None, offset: int | None, @@ -102,9 +110,9 @@ def _perform_search( params: dict[str, Any] = {} final_q = query - # Combine specific convenience query parts if no direct query is provided - if final_q is None and query_parts: - final_q = " AND ".join(query_parts) + # Use the convenience filters if no direct query is provided + if final_q is None: + final_q = built_query if final_q: params["q"] = final_q @@ -162,6 +170,11 @@ def search_proceedings( direct query string or use convenience parameters that will be automatically combined into a query. + Text filters (names, categories, descriptions) are quoted when they + contain spaces or special characters, with inner quotes escaped. Code + filters (trial number, trial type, patent number) that contain spaces + are wrapped in parentheses, so the field applies to every term. + Args: query: Direct query string in USPTO search syntax. sort: Sort order for results. @@ -198,42 +211,28 @@ def search_proceedings( ... limit=50 ... ) """ - q_parts = [] - if trial_number_q: - q_parts.append(f"trialNumber:{trial_number_q}") - if patent_owner_name_q: - q_parts.append(f'patentOwnerData.patentOwnerName:"{patent_owner_name_q}"') - if petitioner_real_party_in_interest_name_q: - q_parts.append( - f'regularPetitionerData.realPartyInInterestName:"{petitioner_real_party_in_interest_name_q}"' - ) - if respondent_name_q: - q_parts.append(f'respondentData.patentOwnerName:"{respondent_name_q}"') - if trial_type_code_q: - q_parts.append(f"trialMetaData.trialTypeCode:{trial_type_code_q}") - if trial_status_category_q: - q_parts.append( - f'trialMetaData.trialStatusCategory:"{trial_status_category_q}"' - ) - - if petition_filing_date_from_q and petition_filing_date_to_q: - q_parts.append( - f"trialMetaData.petitionFilingDate:[{petition_filing_date_from_q} TO {petition_filing_date_to_q}]" - ) - elif petition_filing_date_from_q: - q_parts.append( - f"trialMetaData.petitionFilingDate:>={petition_filing_date_from_q}" - ) - elif petition_filing_date_to_q: - q_parts.append( - f"trialMetaData.petitionFilingDate:<={petition_filing_date_to_q}" - ) + built_query = all_of( + term("trialNumber", trial_number_q), + phrase("patentOwnerData.patentOwnerName", patent_owner_name_q), + phrase( + "regularPetitionerData.realPartyInInterestName", + petitioner_real_party_in_interest_name_q, + ), + phrase("respondentData.patentOwnerName", respondent_name_q), + term("trialMetaData.trialTypeCode", trial_type_code_q), + phrase("trialMetaData.trialStatusCategory", trial_status_category_q), + date_range( + "trialMetaData.petitionFilingDate", + petition_filing_date_from_q, + petition_filing_date_to_q, + ), + ) return self._perform_search( endpoint_key="search_proceedings", response_class=PTABTrialProceedingResponse, query=query, - query_parts=q_parts, + built_query=built_query, post_body=post_body, sort=sort, offset=offset, @@ -276,6 +275,11 @@ def search_documents( direct query string or use convenience parameters that will be automatically combined into a query. + Text filters (names, categories, descriptions) are quoted when they + contain spaces or special characters, with inner quotes escaped. Code + filters (trial number, trial type, patent number) that contain spaces + are wrapped in parentheses, so the field applies to every term. + Args: query: Direct query string in USPTO search syntax. sort: Sort order for results. @@ -314,44 +318,31 @@ def search_documents( ... limit=50 ... ) """ - q_parts = [] - if trial_number_q: - q_parts.append(f"trialNumber:{trial_number_q}") - if document_category_q: - q_parts.append(f'documentData.documentCategory:"{document_category_q}"') - if document_type_name_q: - q_parts.append( - f'documentData.documentTypeDescriptionText:"{document_type_name_q}"' - ) - if petitioner_real_party_in_interest_name_q: - q_parts.append( - f'regularPetitionerData.realPartyInInterestName:"{petitioner_real_party_in_interest_name_q}"' - ) - if inventor_name_q: - q_parts.append(f'patentOwnerData.inventorName:"{inventor_name_q}"') - if real_party_in_interest_name_q: - q_parts.append( - f'regularPetitionerData.realPartyInInterestName:"{real_party_in_interest_name_q}"' - ) - if patent_number_q: - q_parts.append(f"patentOwnerData.patentNumber:{patent_number_q}") - if patent_owner_name_q: - q_parts.append(f'patentOwnerData.patentOwnerName:"{patent_owner_name_q}"') - - if filing_date_from_q and filing_date_to_q: - q_parts.append( - f"documentData.documentFilingDate:[{filing_date_from_q} TO {filing_date_to_q}]" - ) - elif filing_date_from_q: - q_parts.append(f"documentData.documentFilingDate:>={filing_date_from_q}") - elif filing_date_to_q: - q_parts.append(f"documentData.documentFilingDate:<={filing_date_to_q}") + built_query = all_of( + term("trialNumber", trial_number_q), + phrase("documentData.documentCategory", document_category_q), + phrase("documentData.documentTypeDescriptionText", document_type_name_q), + phrase( + "regularPetitionerData.realPartyInInterestName", + petitioner_real_party_in_interest_name_q, + ), + phrase("patentOwnerData.inventorName", inventor_name_q), + phrase( + "regularPetitionerData.realPartyInInterestName", + real_party_in_interest_name_q, + ), + term("patentOwnerData.patentNumber", patent_number_q), + phrase("patentOwnerData.patentOwnerName", patent_owner_name_q), + date_range( + "documentData.documentFilingDate", filing_date_from_q, filing_date_to_q + ), + ) return self._perform_search( endpoint_key="search_documents", response_class=PTABTrialDocumentResponse, query=query, - query_parts=q_parts, + built_query=built_query, post_body=post_body, sort=sort, offset=offset, @@ -396,6 +387,11 @@ def search_decisions( direct query string or use convenience parameters that will be automatically combined into a query. + Text filters (names, categories, descriptions) are quoted when they + contain spaces or special characters, with inner quotes escaped. Code + filters (trial number, trial type, patent number) that contain spaces + are wrapped in parentheses, so the field applies to every term. + Args: query: Direct query string in USPTO search syntax. sort: Sort order for results. @@ -416,6 +412,8 @@ def search_decisions( trial_type_code_q: Filter by trial type code (e.g., "IPR", "PGR", "CBM", "DER"). patent_number_q: Filter by patent number. application_number_q: Filter by application number. + Any common form matches (``17/765,301``, ``17765301``). Write + each number in an ``OR`` or range expression as 8 digits. patent_owner_name_q: Filter by patent owner name. trial_status_category_q: Filter by trial status category. real_party_in_interest_name_q: Filter by real party in interest name. @@ -436,52 +434,38 @@ def search_decisions( ... limit=50 ... ) """ - q_parts = [] - if trial_number_q: - q_parts.append(f"trialNumber:{trial_number_q}") - if decision_type_category_q: - q_parts.append( - f'decisionData.decisionTypeCategory:"{decision_type_category_q}"' - ) - if document_type_description_q: - q_parts.append( - f'documentData.documentTypeDescriptionText:"*{document_type_description_q}*"' - ) - if trial_type_code_q: - q_parts.append(f"trialMetaData.trialTypeCode:{trial_type_code_q}") - if patent_number_q: - q_parts.append(f"patentOwnerData.patentNumber:{patent_number_q}") - if application_number_q: - q_parts.append( - f"patentOwnerData.applicationNumberText:{application_number_q}" - ) - if patent_owner_name_q: - q_parts.append(f'patentOwnerData.patentOwnerName:"{patent_owner_name_q}"') - if trial_status_category_q: - q_parts.append( - f'trialMetaData.trialStatusCategory:"{trial_status_category_q}"' - ) - if real_party_in_interest_name_q: - q_parts.append( - f'regularPetitionerData.realPartyInInterestName:"{real_party_in_interest_name_q}"' - ) - if document_category_q: - q_parts.append(f'documentData.documentCategory:"{document_category_q}"') - - if decision_date_from_q and decision_date_to_q: - q_parts.append( - f"decisionData.decisionIssueDate:[{decision_date_from_q} TO {decision_date_to_q}]" - ) - elif decision_date_from_q: - q_parts.append(f"decisionData.decisionIssueDate:>={decision_date_from_q}") - elif decision_date_to_q: - q_parts.append(f"decisionData.decisionIssueDate:<={decision_date_to_q}") + built_query = all_of( + term("trialNumber", trial_number_q), + phrase("decisionData.decisionTypeCategory", decision_type_category_q), + contains( + "documentData.documentTypeDescriptionText", + document_type_description_q, + quoted=True, + ), + term("trialMetaData.trialTypeCode", trial_type_code_q), + term("patentOwnerData.patentNumber", patent_number_q), + application_number( + "patentOwnerData.applicationNumberText", application_number_q + ), + phrase("patentOwnerData.patentOwnerName", patent_owner_name_q), + phrase("trialMetaData.trialStatusCategory", trial_status_category_q), + phrase( + "regularPetitionerData.realPartyInInterestName", + real_party_in_interest_name_q, + ), + phrase("documentData.documentCategory", document_category_q), + date_range( + "decisionData.decisionIssueDate", + decision_date_from_q, + decision_date_to_q, + ), + ) return self._perform_search( endpoint_key="search_decisions", response_class=PTABTrialDocumentResponse, query=query, - query_parts=q_parts, + built_query=built_query, post_body=post_body, sort=sort, offset=offset, diff --git a/src/pyUSPTO/clients/query.py b/src/pyUSPTO/clients/query.py new file mode 100644 index 0000000..7684567 --- /dev/null +++ b/src/pyUSPTO/clients/query.py @@ -0,0 +1,288 @@ +"""query - Shared builders for Lucene query clauses. + +Each builder returns one ``field:value`` clause, or ``None`` when the filter is +absent (``None`` or ``""``), and formats its value with ``str()``. An empty +list raises ``ValueError``, so a mistaken ``[]`` never runs an unfiltered search. +Text values (``phrase``) are quoted when they contain whitespace or a Lucene +special character, with backslashes and double quotes escaped inside the +quotes. Code values (``term``) are sent as is, inside parentheses when they +contain whitespace, so the field applies to every term of an expression. +""" + +import re + +_PHRASE_SPECIAL = frozenset('+-&|!(){}[]^"~:\\/') + +_PUBLICATION_NUMBER_RE = re.compile( + r"^(?:US)?[\s/\-,]*(\d{4})[\s/\-,]*(\d{7})[\s/\-,]*(?:[A-Z]\d)?$" +) + + +def _is_absent(value: object) -> bool: + """Return True for a filter value that should add no clause.""" + return value is None or value == "" + + +def _has_whitespace(value: str) -> bool: + """Return True if the value contains any whitespace character.""" + return any(c.isspace() for c in value) + + +def sanitize_application_number(input_number: str) -> str: + """Return an application number in the form the USPTO APIs store. + + See ``PatentDataClient.sanitize_application_number`` for the accepted + formats and examples. + + Args: + input_number: Raw application number input. + + Returns: + str: "NNNNNNNN", or "PCT" + 2-char country + 4-digit year + + 6-digit serial. + + Raises: + ValueError: If the format is invalid. + """ + if not input_number or not input_number.strip(): + raise ValueError("Application number cannot be empty") + + raw = input_number.strip() + + # Handle PCT formats. Output is the standardized 15-char form: + # "PCT" + 2-char country + 4-digit year + 6-digit zero-padded serial. + # Example: "PCT/US2024/012345" -> "PCTUS2024012345" + if raw.startswith("PCT"): + parts = raw.split("/") + if len(parts) == 1: + # No slashes — only accept input that is already in the 15-char form. + body = raw[3:] + if len(body) == 12 and body[:2].isalpha() and body[2:].isdigit(): + return raw + raise ValueError( + f"Invalid PCT application format: {input_number}. " + "Expected PCT/CCYYYY/NNNNNN or PCTCCYYYYNNNNNN." + ) + if len(parts) != 3: + raise ValueError( + f"Invalid PCT application format: {input_number}. " + "Expected PCT/CCYYYY/NNNNNN" + ) + + _, country_year, serial = parts + + # country_year can be "US2024" or "US24" + country = country_year[:2] + if len(country) != 2 or not country.isalpha(): + raise ValueError( + f"Invalid PCT country code in: {country_year}. Expected 2 letters." + ) + + year_part = country_year[2:] + if not year_part.isdigit(): + raise ValueError(f"Invalid PCT year in: {country_year}. Must be digits.") + + # Normalize to 4-digit year. 2-digit input uses a sliding window + # tied to the PCT system's start year (1978): YY>=78 -> 19YY, else 20YY. + if len(year_part) == 4: + year = year_part + elif len(year_part) == 2: + yy = int(year_part) + year = f"19{year_part}" if yy >= 78 else f"20{year_part}" + else: + raise ValueError( + f"Invalid PCT year length in: {country_year}. Expected CCYYYY or CCYY." + ) + + # Serial must be digits only, and fit in 6 digits. + if not serial.isdigit(): + raise ValueError(f"Invalid PCT serial: {serial}. Must be numeric.") + if len(serial) > 6: + raise ValueError(f"Invalid PCT serial: {serial}. Must be at most 6 digits.") + + return f"PCT{country}{year}{serial.zfill(6)}" + + # Strip whitespace and remove commas/spaces + cleaned = raw.replace(",", "").replace(" ", "") + + # Check if this is series code format (NN/NNNNNN) + if "/" in cleaned: + parts = cleaned.split("/") + if len(parts) != 2: + raise ValueError( + f"Invalid application number format: {input_number}. " + "Expected format: NNNNNNNN or NN/NNNNNN" + ) + + series, serial = parts + if not series.isdigit() or not serial.isdigit(): + raise ValueError( + f"Invalid application number format: {input_number}. " + "Series and serial must be numeric." + ) + + if len(series) != 2 or len(serial) != 6: + raise ValueError( + f"Invalid application number format: {input_number}. " + "Expected series code format: NN/NNNNNN (2 digits / 6 digits)" + ) + + return series + serial + + # Standard 8-digit format + if not cleaned.isdigit(): + raise ValueError( + f"Invalid application number format: {input_number}. " + "Must contain only digits." + ) + + if len(cleaned) != 8: + raise ValueError( + f"Invalid application number format: {input_number}. Expected 8 digits." + ) + + return cleaned + + +def _application_number_value(value: object) -> str: + """Return the search value for an application-number field. + + Values that look like an application number are sanitized, so + ``14/412,875`` and ``14412875`` search the same way. Values that contain a + wildcard, or that the sanitizer rejects (such as range or ``OR`` + expressions), are returned inside parentheses, so the field applies to + every term of the expression. + """ + text = str(value) + if "*" not in text: + try: + return sanitize_application_number(text) + except ValueError: + pass + return f"({text})" + + +def application_number(field: str, value: object) -> str | None: + """Build a clause for an application-number field.""" + if _is_absent(value): + return None + return f"{field}:{_application_number_value(value)}" + + +def publication_number(field: str, value: object) -> str | None: + """Build a clause for a US publication-number field. + + The API stores publication numbers as ``USyyyyNNNNNNNAn`` (e.g. + ``US20210138233A1``). The value is reduced to its 11 digits (year + + 7-digit serial) inside wildcards, so any prefix or kind code matches. + Values that already contain a wildcard, or that do not look like a US + publication number, are sent unchanged. + """ + if _is_absent(value): + return None + text = str(value) + if "*" in text: + return f"{field}:{text}" + match = _PUBLICATION_NUMBER_RE.match(text.strip().upper()) + if match is None: + return f"{field}:{text}" + return f"{field}:*{match.group(1)}{match.group(2)}*" + + +def term(field: str, value: object) -> str | None: + """Build a clause for a code or number field. + + A value containing whitespace is wrapped in parentheses, so the field + applies to every term of an expression such as ``1700 OR 2800``. + """ + if _is_absent(value): + return None + text = str(value) + if _has_whitespace(text): + return f"{field}:({text})" + return f"{field}:{text}" + + +def phrase(field: str, value: object) -> str | None: + """Build a clause for a text field (names, titles, categories). + + A single plain word is sent as is, so wildcards such as ``Smith*`` still + work. Any other value is quoted, with backslashes and double quotes + escaped. A wildcard inside a quoted value is searched as plain text, so + ``Smith-K*`` does not expand. + """ + if _is_absent(value): + return None + text = str(value) + if not _has_whitespace(text) and not any(c in _PHRASE_SPECIAL for c in text): + return f"{field}:{text}" + escaped = text.replace("\\", "\\\\").replace('"', '\\"') + return f'{field}:"{escaped}"' + + +def contains(field: str, value: object, *, quoted: bool = False) -> str | None: + """Build a clause that matches the value anywhere in the field. + + With ``quoted``, backslashes and double quotes in the value are escaped. + """ + if _is_absent(value): + return None + if quoted: + escaped = str(value).replace("\\", "\\\\").replace('"', '\\"') + return f'{field}:"*{escaped}*"' + return f"{field}:*{value}*" + + +def any_of(field: str, value: object) -> str | None: + """Build ``field:(a OR b)`` for a list, or a ``term`` for one value. + + Raises: + ValueError: If the value is an empty list. + """ + if _is_absent(value): + return None + if isinstance(value, list): + if not value: + raise ValueError( + f"{field} filter got an empty list; pass None to skip the filter." + ) + return f"{field}:({' OR '.join(str(v) for v in value)})" + return term(field, value) + + +def boolean(field: str, value: bool | None) -> str | None: + """Build ``field:true`` or ``field:false``.""" + if value is None: + return None + return f"{field}:{str(value).lower()}" + + +def date_range(field: str, start: object, end: object) -> str | None: + """Build a date range clause from optional start and end dates.""" + has_start = not _is_absent(start) + has_end = not _is_absent(end) + if has_start and has_end: + return f"{field}:[{start} TO {end}]" + if has_start: + return f"{field}:>={start}" + if has_end: + return f"{field}:<={end}" + return None + + +def either(*clauses: str | None) -> str | None: + """Join the present clauses with OR inside parentheses.""" + present = [c for c in clauses if c] + if not present: + return None + if len(present) == 1: + return present[0] + return f"({' OR '.join(present)})" + + +def all_of(*clauses: str | None) -> str | None: + """Join the present clauses with AND.""" + present = [c for c in clauses if c] + if not present: + return None + return " AND ".join(present) diff --git a/src/pyUSPTO/exceptions.py b/src/pyUSPTO/exceptions.py index 51f5943..7701428 100644 --- a/src/pyUSPTO/exceptions.py +++ b/src/pyUSPTO/exceptions.py @@ -14,6 +14,14 @@ from pyUSPTO.models.patent_data import Document +# The API gateway answers a request for a URL path that does not exist with +# HTTP 403 and this body, so the 403 alone does not prove an auth problem. +_MISSING_AUTH_TOKEN_HINT = ( + "USPTO also returns 'Missing Authentication Token' " + "when the requested URL path does not exist" +) + + # --- Exception Classes --- class USPTOApiError(Exception): """Base exception for USPTO API errors. @@ -252,8 +260,16 @@ def from_http_error( ): # If it might have been JSON but parsing failed error_details_from_response = http_error.response.text + message = client_operation_message + if ( + status_code == 403 + and isinstance(error_data, dict) + and error_data.get("message") == "Missing Authentication Token" + ): + message = f"{client_operation_message} ({_MISSING_AUTH_TOKEN_HINT})" + return cls( - message=client_operation_message, + message=message, status_code=status_code, api_short_error=api_short_error_from_response, error_details=error_details_from_response, diff --git a/tests/clients/test_base.py b/tests/clients/test_base.py index bbdaeb0..834b17b 100644 --- a/tests/clients/test_base.py +++ b/tests/clients/test_base.py @@ -4,6 +4,7 @@ This module contains tests for the BaseUSPTOClient class. """ +import uuid from typing import Any, cast from unittest.mock import MagicMock, mock_open, patch @@ -104,6 +105,24 @@ def from_dict( class TestBaseUSPTOClient: """Tests for the BaseUSPTOClient class.""" + def test_quote_path_value(self) -> None: + """Test URL path values are percent-encoded, plain IDs unchanged.""" + assert BaseUSPTOClient._quote_path_value("abc?x/y z#1") == ( + "abc%3Fx%2Fy%20z%231" + ) + assert BaseUSPTOClient._quote_path_value( + "9f1a4a2b-eee1-58ec-a3aa-167c4075aed4" + ) == ("9f1a4a2b-eee1-58ec-a3aa-167c4075aed4") + assert BaseUSPTOClient._quote_path_value("ipa250102.zip") == "ipa250102.zip" + + def test_quote_path_value_non_string(self) -> None: + """Test non-string IDs are formatted with str() before encoding.""" + decision_id = uuid.UUID("9f1a4a2b-eee1-58ec-a3aa-167c4075aed4") + assert BaseUSPTOClient._quote_path_value(decision_id) == ( + "9f1a4a2b-eee1-58ec-a3aa-167c4075aed4" + ) + assert BaseUSPTOClient._quote_path_value(12345) == "12345" + def test_init(self) -> None: """Test initialization of the BaseUSPTOClient.""" # Test with API key diff --git a/tests/clients/test_bulk_data_clients.py b/tests/clients/test_bulk_data_clients.py index 97e7670..1aa6bbe 100644 --- a/tests/clients/test_bulk_data_clients.py +++ b/tests/clients/test_bulk_data_clients.py @@ -5,6 +5,7 @@ model handling, edge cases, and response handling. """ +import warnings from datetime import UTC, date, datetime from typing import Any from unittest.mock import MagicMock, call, patch @@ -19,6 +20,7 @@ FileData, ProductFileBag, ) +from pyUSPTO.warnings import USPTODataMismatchWarning # --- Fixtures --- @@ -340,6 +342,58 @@ def test_get_product_by_id( assert isinstance(product, BulkDataProduct) assert product.product_identifier == "PRODUCT1" + def test_get_product_by_id_escapes_reserved_characters( + self, mock_bulk_data_client: BulkDataClient, bulk_data_sample: dict[str, Any] + ) -> None: + """Test '/' and '?' in a product ID are percent-encoded in the path.""" + mock_response = MagicMock() + mock_response.json.return_value = bulk_data_sample + mock_session = MagicMock() + mock_session.get.return_value = mock_response + mock_bulk_data_client.config._session = mock_session + + # The sample's product has a different ID, so a mismatch warning fires. + with warnings.catch_warnings(): + warnings.simplefilter("ignore", USPTODataMismatchWarning) + mock_bulk_data_client.get_product_by_id(product_id="A/B?C") + + assert mock_session.get.call_args.kwargs["url"] == ( + f"{mock_bulk_data_client.base_url}/api/v1/datasets/products/A%2FB%3FC" + ) + + def test_download_file_escapes_file_name( + self, mock_bulk_data_client: BulkDataClient + ) -> None: + """Test the fallback download URL percent-encodes the file name.""" + file_data = FileData( + file_name="test file#1.zip", + file_size=1024, + product_identifier="PRODUCT1", + file_data_from_date=date(2023, 1, 1), + file_data_to_date=date(2023, 12, 31), + file_type_text="ZIP", + file_release_date=datetime(2024, 1, 1, tzinfo=UTC), + ) + + with patch.object( + mock_bulk_data_client, + "_download_file", + return_value="./downloads/test file#1.zip", + ) as mock_download: + mock_bulk_data_client.download_file( + file_data=file_data, destination="./downloads" + ) + + mock_download.assert_called_once_with( + url=( + f"{mock_bulk_data_client.base_url}/api/v1/datasets/products/" + "files/PRODUCT1/test%20file%231.zip" + ), + destination="./downloads", + file_name="test file#1.zip", + overwrite=False, + ) + def test_download_file(self, mock_bulk_data_client: BulkDataClient) -> None: """Test download_file method.""" # Setup @@ -569,8 +623,8 @@ def test_download_file_with_custom_filename( extract=False, ) - # Verify custom filename is used - expected_url = f"{mock_bulk_data_client.base_url}/api/v1/datasets/products/files/PRODUCT1/custom.zip" + # The URL requests the real file; the custom name is only the local save name + expected_url = f"{mock_bulk_data_client.base_url}/api/v1/datasets/products/files/PRODUCT1/original.zip" mock_download.assert_called_once_with( url=expected_url, destination=destination, diff --git a/tests/clients/test_enriched_citation_clients.py b/tests/clients/test_enriched_citation_clients.py index 7fc7f2a..4f1265a 100644 --- a/tests/clients/test_enriched_citation_clients.py +++ b/tests/clients/test_enriched_citation_clients.py @@ -205,6 +205,34 @@ def test_search_with_application_number( json_data = call_args[1]["json_data"] assert "patentApplicationNumber:15739603" in json_data["criteria"] + def test_search_with_office_action_category_or_expression( + self, + client_with_mocked_request: tuple[EnrichedCitationsClient, MagicMock], + mock_enriched_response_with_data: EnrichedCitationResponse, + ) -> None: + """Test an OR expression in office_action_category_q is wrapped.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_enriched_response_with_data + + client.search_citations(office_action_category_q="CTNF OR CTFR") + + json_data = mock_get_model.call_args[1]["json_data"] + assert json_data["criteria"] == "officeActionCategory:(CTNF OR CTFR)" + + def test_search_with_series_code_application_number( + self, + client_with_mocked_request: tuple[EnrichedCitationsClient, MagicMock], + mock_enriched_response_with_data: EnrichedCitationResponse, + ) -> None: + """Test a series-code patent_application_number_q is sent as 8 digits.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_enriched_response_with_data + + client.search_citations(patent_application_number_q="15/739,603") + + json_data = mock_get_model.call_args[1]["json_data"] + assert json_data["criteria"] == "patentApplicationNumber:15739603" + def test_search_with_cited_document_identifier( self, client_with_mocked_request: tuple[EnrichedCitationsClient, MagicMock], diff --git a/tests/clients/test_oa_actions_clients.py b/tests/clients/test_oa_actions_clients.py index e408d53..0ae3c9f 100644 --- a/tests/clients/test_oa_actions_clients.py +++ b/tests/clients/test_oa_actions_clients.py @@ -179,6 +179,35 @@ def test_patent_application_number_q( call_kwargs["json_data"]["criteria"] == "patentApplicationNumber:14485382" ) + def test_group_art_unit_number_q_empty_string( + self, + client_with_mocked_request: tuple[OAActionsClient, MagicMock], + mock_response_with_data: OAActionsResponse, + ) -> None: + """Test an empty group_art_unit_number_q adds no clause.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(group_art_unit_number_q="", tech_center_q="1700") + + call_kwargs = mock_get_model.call_args.kwargs + assert call_kwargs["json_data"]["criteria"] == "techCenter:1700" + + def test_patent_application_number_q_series_code( + self, + client_with_mocked_request: tuple[OAActionsClient, MagicMock], + mock_response_with_data: OAActionsResponse, + ) -> None: + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(patent_application_number_q="14/485,382") + + call_kwargs = mock_get_model.call_args.kwargs + assert ( + call_kwargs["json_data"]["criteria"] == "patentApplicationNumber:14485382" + ) + def test_legacy_document_code_identifier_q( self, client_with_mocked_request: tuple[OAActionsClient, MagicMock], @@ -246,6 +275,40 @@ def test_application_type_category_q( call_kwargs = mock_get_model.call_args.kwargs assert call_kwargs["json_data"]["criteria"] == "applicationTypeCategory:REGULAR" + def test_application_type_category_q_or_expression( + self, + client_with_mocked_request: tuple[OAActionsClient, MagicMock], + mock_response_with_data: OAActionsResponse, + ) -> None: + """Test an OR expression in application_type_category_q is wrapped.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(application_type_category_q="REGULAR OR Y") + + call_kwargs = mock_get_model.call_args.kwargs + assert ( + call_kwargs["json_data"]["criteria"] + == "applicationTypeCategory:(REGULAR OR Y)" + ) + + def test_access_level_category_q_or_expression( + self, + client_with_mocked_request: tuple[OAActionsClient, MagicMock], + mock_response_with_data: OAActionsResponse, + ) -> None: + """Test an OR expression in access_level_category_q is wrapped.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(access_level_category_q="PUBLIC OR INTERNAL") + + call_kwargs = mock_get_model.call_args.kwargs + assert ( + call_kwargs["json_data"]["criteria"] + == "accessLevelCategory:(PUBLIC OR INTERNAL)" + ) + def test_submission_date_range( self, client_with_mocked_request: tuple[OAActionsClient, MagicMock], diff --git a/tests/clients/test_oa_citations_clients.py b/tests/clients/test_oa_citations_clients.py index a90cb93..d3e1e77 100644 --- a/tests/clients/test_oa_citations_clients.py +++ b/tests/clients/test_oa_citations_clients.py @@ -185,6 +185,21 @@ def test_patent_application_number_q( call_kwargs["json_data"]["criteria"] == "patentApplicationNumber:17519936" ) + def test_patent_application_number_q_series_code( + self, + client_with_mocked_request: tuple[OACitationsClient, MagicMock], + mock_response_with_data: OACitationsResponse, + ) -> None: + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(patent_application_number_q="17/519,936") + + call_kwargs = mock_get_model.call_args.kwargs + assert ( + call_kwargs["json_data"]["criteria"] == "patentApplicationNumber:17519936" + ) + def test_legal_section_code_q( self, client_with_mocked_request: tuple[OACitationsClient, MagicMock], @@ -211,6 +226,23 @@ def test_action_type_category_q( call_kwargs = mock_get_model.call_args.kwargs assert call_kwargs["json_data"]["criteria"] == "actionTypeCategory:rejected" + def test_action_type_category_q_or_expression( + self, + client_with_mocked_request: tuple[OACitationsClient, MagicMock], + mock_response_with_data: OACitationsResponse, + ) -> None: + """Test an OR expression in action_type_category_q is wrapped.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(action_type_category_q="rejected OR objected") + + call_kwargs = mock_get_model.call_args.kwargs + assert ( + call_kwargs["json_data"]["criteria"] + == "actionTypeCategory:(rejected OR objected)" + ) + def test_tech_center_q( self, client_with_mocked_request: tuple[OACitationsClient, MagicMock], diff --git a/tests/clients/test_oa_rejections_clients.py b/tests/clients/test_oa_rejections_clients.py index 93cc1c6..603b936 100644 --- a/tests/clients/test_oa_rejections_clients.py +++ b/tests/clients/test_oa_rejections_clients.py @@ -183,6 +183,21 @@ def test_patent_application_number_q( call_kwargs["json_data"]["criteria"] == "patentApplicationNumber:12190351" ) + def test_patent_application_number_q_series_code( + self, + client_with_mocked_request: tuple[OARejectionsClient, MagicMock], + mock_response_with_data: OARejectionsResponse, + ) -> None: + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(patent_application_number_q="12/190,351") + + call_kwargs = mock_get_model.call_args.kwargs + assert ( + call_kwargs["json_data"]["criteria"] == "patentApplicationNumber:12190351" + ) + def test_legacy_document_code_identifier_q( self, client_with_mocked_request: tuple[OARejectionsClient, MagicMock], @@ -237,6 +252,23 @@ def test_action_type_category_q( call_kwargs = mock_get_model.call_args.kwargs assert call_kwargs["json_data"]["criteria"] == "actionTypeCategory:rejected" + def test_action_type_category_q_or_expression( + self, + client_with_mocked_request: tuple[OARejectionsClient, MagicMock], + mock_response_with_data: OARejectionsResponse, + ) -> None: + """Test an OR expression in action_type_category_q is wrapped.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_response_with_data + + client.search(action_type_category_q="rejected OR objected") + + call_kwargs = mock_get_model.call_args.kwargs + assert ( + call_kwargs["json_data"]["criteria"] + == "actionTypeCategory:(rejected OR objected)" + ) + def test_submission_date_range( self, client_with_mocked_request: tuple[OARejectionsClient, MagicMock], diff --git a/tests/clients/test_patent_data_clients.py b/tests/clients/test_patent_data_clients.py index 3f0c3c0..73c89a7 100644 --- a/tests/clients/test_patent_data_clients.py +++ b/tests/clients/test_patent_data_clients.py @@ -8,6 +8,7 @@ import csv import io import os +import warnings from collections.abc import Iterator from datetime import UTC, date, datetime from typing import Any @@ -307,6 +308,20 @@ def test_init_without_config(self, monkeypatch: Any) -> None: class TestPatentApplicationSearch: """Tests for patent application search functionalities using the new search_applications method.""" + @pytest.mark.parametrize("param", ["customer_number_q", "status_code_q"]) + def test_search_applications_empty_list_raises( + self, + param: str, + client_with_mocked_request: tuple[PatentDataClient, MagicMock], + ) -> None: + """Test an empty list filter raises instead of searching unfiltered.""" + client, mock_get_model = client_with_mocked_request + + with pytest.raises(ValueError, match="empty list"): + client.search_applications(**{param: []}) + + mock_get_model.assert_not_called() + def test_search_applications_get_direct_query( self, client_with_mocked_request: tuple[PatentDataClient, MagicMock], @@ -384,8 +399,41 @@ def test_search_applications_post( @pytest.mark.parametrize( "search_q_params, expected_q_part", [ - ({"application_number_q": "app123"}, "applicationNumberText:app123"), + ({"application_number_q": "app123"}, "applicationNumberText:(app123)"), + ( + {"application_number_q": "14/412,875"}, + "applicationNumberText:14412875", + ), + ( + {"application_number_q": "14/412875"}, + "applicationNumberText:14412875", + ), + ( + {"application_number_q": 14412875}, + "applicationNumberText:14412875", + ), + ({"application_number_q": "16*"}, "applicationNumberText:(16*)"), + ( + {"application_number_q": "14412875 OR 14412876"}, + "applicationNumberText:(14412875 OR 14412876)", + ), + ( + {"application_number_q": "[16000000 TO 16999999]"}, + "applicationNumberText:([16000000 TO 16999999])", + ), + ( + {"application_number_q": "PCT/US2024/012345"}, + "applicationNumberText:PCTUS2024012345", + ), ({"patent_number_q": "PN123"}, "applicationMetaData.patentNumber:PN123"), + ( + {"patent_number_q": "10000000 OR 10000001"}, + "applicationMetaData.patentNumber:(10000000 OR 10000001)", + ), + ( + {"inventor_name_q": 'Smith "John'}, + 'applicationMetaData.inventorBag.inventorNameText:"Smith \\"John"', + ), ( {"inventor_name_q": "Doe J"}, 'applicationMetaData.inventorBag.inventorNameText:"Doe J"', @@ -820,6 +868,27 @@ def test_get_application_by_number_success( assert result.application_meta_data is not None assert result.application_meta_data.invention_title == "Test Invention" + def test_get_application_by_number_series_code_input( + self, + client_with_mocked_request: tuple[PatentDataClient, MagicMock], + mock_patent_data_response_with_data: PatentDataResponse, + mock_patent_file_wrapper: PatentFileWrapper, + ) -> None: + """Test a series-code number requests the 8-digit path without a mismatch warning.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_patent_data_response_with_data + + with warnings.catch_warnings(): + warnings.simplefilter("error", USPTODataMismatchWarning) + result = client.get_application_by_number(application_number="12/345,678") + + mock_get_model.assert_called_once_with( + method="GET", + endpoint="api/v1/patent/applications/12345678", + response_class=PatentDataResponse, + ) + assert result is mock_patent_file_wrapper + def test_get_application_by_number_empty_bag_returns_none( self, client_with_mocked_request: tuple[PatentDataClient, MagicMock], @@ -2509,8 +2578,41 @@ def test_get_search_results_get_with_combined_q_convenience_params( @pytest.mark.parametrize( "search_q_params, expected_q_part", [ - ({"application_number_q": "app123"}, "applicationNumberText:app123"), + ({"application_number_q": "app123"}, "applicationNumberText:(app123)"), + ( + {"application_number_q": "14/412,875"}, + "applicationNumberText:14412875", + ), + ( + {"application_number_q": "14/412875"}, + "applicationNumberText:14412875", + ), + ( + {"application_number_q": 14412875}, + "applicationNumberText:14412875", + ), + ({"application_number_q": "16*"}, "applicationNumberText:(16*)"), + ( + {"application_number_q": "14412875 OR 14412876"}, + "applicationNumberText:(14412875 OR 14412876)", + ), + ( + {"application_number_q": "[16000000 TO 16999999]"}, + "applicationNumberText:([16000000 TO 16999999])", + ), + ( + {"application_number_q": "PCT/US2024/012345"}, + "applicationNumberText:PCTUS2024012345", + ), ({"patent_number_q": "PN123"}, "applicationMetaData.patentNumber:PN123"), + ( + {"patent_number_q": "10000000 OR 10000001"}, + "applicationMetaData.patentNumber:(10000000 OR 10000001)", + ), + ( + {"inventor_name_q": 'Smith "John'}, + 'applicationMetaData.inventorBag.inventorNameText:"Smith \\"John"', + ), ( {"inventor_name_q": "Doe J"}, 'applicationMetaData.inventorBag.inventorNameText:"Doe J"', @@ -3024,21 +3126,19 @@ def test_sanitize_with_spaces(self, patent_data_client: PatentDataClient) -> Non def test_sanitize_series_code_format( self, patent_data_client: PatentDataClient ) -> None: - """Test series code format (NN/NNNNNN).""" - assert ( - patent_data_client.sanitize_application_number("08/123456") == "08/123456" - ) + """Test series code format (NN/NNNNNN) is joined into 8 digits.""" + assert patent_data_client.sanitize_application_number("08/123456") == "08123456" def test_sanitize_series_code_with_separators( self, patent_data_client: PatentDataClient ) -> None: """Test series code format with commas and spaces.""" assert ( - patent_data_client.sanitize_application_number("08/123,456") == "08/123456" + patent_data_client.sanitize_application_number("08/123,456") == "08123456" ) assert ( patent_data_client.sanitize_application_number(" 08 / 123 456 ") - == "08/123456" + == "08123456" ) def test_sanitize_empty_string_raises( diff --git a/tests/clients/test_petition_decision_clients.py b/tests/clients/test_petition_decision_clients.py index e2904a3..ca0b348 100644 --- a/tests/clients/test_petition_decision_clients.py +++ b/tests/clients/test_petition_decision_clients.py @@ -6,6 +6,7 @@ that restricts it to final decisions. """ +import warnings from collections.abc import Iterator from typing import Any from unittest.mock import MagicMock, Mock, call, patch @@ -243,6 +244,34 @@ def test_search_with_application_number( ) assert result is mock_petition_response_with_data + def test_search_with_quote_in_inventor_name( + self, + client_with_mocked_request: tuple[PetitionDecisionsClient, MagicMock], + mock_petition_response_with_data: PetitionDecisionResponse, + ) -> None: + """Test a double quote inside inventor_name_q is escaped.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_petition_response_with_data + + client.search_decisions(inventor_name_q='Smith "John', limit=25) + + params = mock_get_model.call_args[1]["params"] + assert params["q"] == 'inventorBag:"Smith \\"John"' + + def test_search_with_series_code_application_number( + self, + client_with_mocked_request: tuple[PetitionDecisionsClient, MagicMock], + mock_petition_response_with_data: PetitionDecisionResponse, + ) -> None: + """Test a series-code application_number_q is sent as 8 digits.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_petition_response_with_data + + client.search_decisions(application_number_q="17/765,301", limit=25) + + params = mock_get_model.call_args[1]["params"] + assert params["q"] == "applicationNumberText:17765301" + def test_search_with_date_range( self, client_with_mocked_request: tuple[PetitionDecisionsClient, MagicMock], @@ -523,6 +552,27 @@ def test_get_decision_by_id( assert result is not None assert result.petition_decision_record_identifier == record_id + def test_get_decision_by_id_escapes_reserved_characters( + self, + client_with_mocked_request: tuple[PetitionDecisionsClient, MagicMock], + mock_petition_response_with_data: PetitionDecisionResponse, + ) -> None: + """Test '?' and '/' in a decision ID are percent-encoded in the path.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = mock_petition_response_with_data + + # The fixture's decision has a different ID, so a mismatch warning fires. + with warnings.catch_warnings(): + warnings.simplefilter("ignore", USPTODataMismatchWarning) + client.get_decision_by_id("abc?x/y") + + mock_get_model.assert_called_once_with( + method="GET", + endpoint="api/v1/petition/decisions/abc%3Fx%2Fy", + params=None, + response_class=PetitionDecisionResponse, + ) + def test_get_decision_by_id_not_found( self, client_with_mocked_request: tuple[PetitionDecisionsClient, MagicMock], @@ -653,6 +703,21 @@ def test_download_with_application_number( params = call_args[1]["params"] assert "applicationNumberText:17765301" in params["q"] + def test_download_with_series_code_application_number( + self, + client_with_mocked_request: tuple[PetitionDecisionsClient, MagicMock], + ) -> None: + """Test a series-code application_number_q is sent as 8 digits.""" + client, mock_get_model = client_with_mocked_request + mock_get_model.return_value = PetitionDecisionDownloadResponse( + petition_decision_data=[] + ) + + client.download_decisions(format="json", application_number_q="17/765301") + + params = mock_get_model.call_args[1]["params"] + assert params["q"] == "applicationNumberText:17765301" + def test_download_with_inventor_name( self, client_with_mocked_request: tuple[PetitionDecisionsClient, MagicMock], diff --git a/tests/clients/test_ptab_appeals_client.py b/tests/clients/test_ptab_appeals_client.py index dd93fa3..c83177e 100644 --- a/tests/clients/test_ptab_appeals_client.py +++ b/tests/clients/test_ptab_appeals_client.py @@ -237,9 +237,9 @@ def test_search_decisions_get_with_all_convenience_params( call_args = mock_session.get.call_args params = call_args[1]["params"] assert "appealNumber:2023-001234" in params["q"] - assert "applicationNumberText:15/123456" in params["q"] - assert "appellantData.realPartyInInterestName:Test Appellant" in params["q"] - assert "appellantData.counselName:Test Requestor" in params["q"] + assert "appellantData.applicationNumberText:15123456" in params["q"] + assert 'appellantData.realPartyInInterestName:"Test Appellant"' in params["q"] + assert 'appellantData.counselName:"Test Requestor"' in params["q"] assert "decisionTypeCategory:Affirmed" in params["q"] assert "technologyCenterNumber:3600" in params["q"] diff --git a/tests/clients/test_ptab_interferences_client.py b/tests/clients/test_ptab_interferences_client.py index 786a612..bc954f6 100644 --- a/tests/clients/test_ptab_interferences_client.py +++ b/tests/clients/test_ptab_interferences_client.py @@ -217,7 +217,7 @@ def test_search_decisions_get_with_convenience_params( assert ( 'documentData.interferenceOutcomeCategory:"Final Decision"' in params["q"] ) - assert 'documentData.decisionTypeCategory:"Decision"' in params["q"] + assert "documentData.decisionTypeCategory:Decision" in params["q"] assert ( "documentData.decisionIssueDate:[2023-01-01 TO 2023-12-31]" in params["q"] ) @@ -302,10 +302,16 @@ def test_search_decisions_get_with_all_convenience_params( call_args = mock_session.get.call_args params = call_args[1]["params"] assert "interferenceNumber:106123" in params["q"] - assert "seniorPartyData.applicationNumberText:12/345678" in params["q"] - assert "juniorPartyData.applicationNumberText:13/987654" in params["q"] + assert "seniorPartyData.applicationNumberText:12345678" in params["q"] + assert "juniorPartyData.applicationNumberText:13987654" in params["q"] assert 'seniorPartyData.patentOwnerName:"Senior Party Inc."' in params["q"] assert 'juniorPartyData.patentOwnerName:"Junior Party LLC"' in params["q"] + assert ( + '(seniorPartyData.patentOwnerName:"Senior Party Inc."' + ' OR seniorPartyData.inventorName:"Senior Party Inc."' + ' OR seniorPartyData.realPartyInInterestName:"Senior Party Inc.")' + in params["q"] + ) assert ( 'documentData.interferenceOutcomeCategory:"Priority to Senior Party"' in params["q"] diff --git a/tests/clients/test_ptab_trials_client.py b/tests/clients/test_ptab_trials_client.py index df40888..e8e6a63 100644 --- a/tests/clients/test_ptab_trials_client.py +++ b/tests/clients/test_ptab_trials_client.py @@ -228,7 +228,7 @@ def test_search_proceedings_with_all_convenience_params( in params["q"] ) assert 'respondentData.patentOwnerName:"Test Respondent"' in params["q"] - assert 'trialMetaData.trialStatusCategory:"Instituted"' in params["q"] + assert "trialMetaData.trialStatusCategory:Instituted" in params["q"] def test_search_proceedings_with_date_from_only( self, @@ -430,7 +430,7 @@ def test_search_documents_with_all_convenience_params( call_args = mock_session.get.call_args params = call_args[1]["params"] assert "trialNumber:IPR2023-00001" in params["q"] - assert 'documentData.documentCategory:"Paper"' in params["q"] + assert "documentData.documentCategory:Paper" in params["q"] assert ( 'documentData.documentTypeDescriptionText:"Patent Owner Response"' in params["q"] @@ -649,14 +649,14 @@ def test_search_decisions_with_all_convenience_params( ) assert "trialTypeCode:IPR" in params["q"] assert "patentOwnerData.patentNumber:US1234567" in params["q"] - assert "patentOwnerData.applicationNumberText:15/123456" in params["q"] + assert "patentOwnerData.applicationNumberText:15123456" in params["q"] assert 'patentOwnerData.patentOwnerName:"Test Owner"' in params["q"] - assert 'trialMetaData.trialStatusCategory:"Instituted"' in params["q"] + assert "trialMetaData.trialStatusCategory:Instituted" in params["q"] assert ( 'regularPetitionerData.realPartyInInterestName:"Real Party LLC"' in params["q"] ) - assert 'documentData.documentCategory:"Decision"' in params["q"] + assert "documentData.documentCategory:Decision" in params["q"] assert ( "decisionData.decisionIssueDate:[2023-01-01 TO 2023-12-31]" in params["q"] ) diff --git a/tests/clients/test_query.py b/tests/clients/test_query.py new file mode 100644 index 0000000..591df05 --- /dev/null +++ b/tests/clients/test_query.py @@ -0,0 +1,148 @@ +"""Tests for the shared query clause builders in pyUSPTO.clients.query.""" + +from typing import Any + +import pytest + +from pyUSPTO.clients import query + + +@pytest.mark.parametrize( + "value, expected", + [ + ("CTNF", "f:CTNF"), + (1700, "f:1700"), + ("16*", "f:16*"), + ("1700 OR 2800", "f:(1700 OR 2800)"), + (None, None), + ("", None), + ], +) +def test_term(value: Any, expected: str | None) -> None: + """Test code values are sent raw, and expressions are wrapped.""" + assert query.term("f", value) == expected + + +@pytest.mark.parametrize( + "value, expected", + [ + ("Smith", "f:Smith"), + ("Smith*", "f:Smith*"), + ("John Smith", 'f:"John Smith"'), + ("Block, Inc.", 'f:"Block, Inc."'), + ("H01L21/02", 'f:"H01L21/02"'), + ("E-Link", 'f:"E-Link"'), + ("Smith-K*", 'f:"Smith-K*"'), + ('Smith "John', 'f:"Smith \\"John"'), + ("a\\b", 'f:"a\\\\b"'), + ('"John Smith"', 'f:"\\"John Smith\\""'), + (None, None), + ("", None), + ], +) +def test_phrase(value: Any, expected: str | None) -> None: + """Test text values are quoted when needed, with quotes escaped.""" + assert query.phrase("f", value) == expected + + +def test_contains() -> None: + """Test contains wraps the value in wildcards.""" + assert query.contains("f", "103") == "f:*103*" + assert query.contains("f", "Final", quoted=True) == 'f:"*Final*"' + assert query.contains("f", None) is None + + +def test_contains_quoted_escapes() -> None: + """Test a quoted contains clause escapes backslashes and double quotes.""" + assert query.contains("f", 'Final "Written', quoted=True) == ( + 'f:"*Final \\"Written*"' + ) + assert query.contains("f", "a\\b", quoted=True) == 'f:"*a\\\\b*"' + + +@pytest.mark.parametrize( + "value, expected", + [ + ([1, "2"], "f:(1 OR 2)"), + (150, "f:150"), + ("150 OR 161", "f:(150 OR 161)"), + (None, None), + ], +) +def test_any_of(value: Any, expected: str | None) -> None: + """Test lists become an OR group and single values a term.""" + assert query.any_of("f", value) == expected + + +def test_any_of_empty_list() -> None: + """Test an empty list raises instead of silently dropping the filter.""" + with pytest.raises(ValueError, match="empty list"): + query.any_of("f", []) + + +@pytest.mark.parametrize( + "value, expected", [(True, "f:true"), (False, "f:false"), (None, None)] +) +def test_boolean(value: bool | None, expected: str | None) -> None: + """Test booleans are lowercased, and False still adds a clause.""" + assert query.boolean("f", value) == expected + + +@pytest.mark.parametrize( + "start, end, expected", + [ + ("2023-01-01", "2023-12-31", "f:[2023-01-01 TO 2023-12-31]"), + ("2023-01-01", None, "f:>=2023-01-01"), + (None, "2023-12-31", "f:<=2023-12-31"), + (None, None, None), + ], +) +def test_date_range(start: str | None, end: str | None, expected: str | None) -> None: + """Test the three date range forms.""" + assert query.date_range("f", start, end) == expected + + +@pytest.mark.parametrize( + "value, expected", + [ + ("14/412,875", "f:14412875"), + ("14/412875", "f:14412875"), + (14412875, "f:14412875"), + ("PCT/US2024/012345", "f:PCTUS2024012345"), + ("16*", "f:(16*)"), + ("14412875 OR 14412876", "f:(14412875 OR 14412876)"), + ("[16000000 TO 16999999]", "f:([16000000 TO 16999999])"), + (None, None), + ], +) +def test_application_number(value: Any, expected: str | None) -> None: + """Test application numbers are sanitized, and expressions wrapped.""" + assert query.application_number("f", value) == expected + + +@pytest.mark.parametrize( + "value, expected", + [ + ("2021/0138233", "f:*20210138233*"), + ("US 2021/0138233 A1", "f:*20210138233*"), + ("US2021*", "f:US2021*"), + ("abc", "f:abc"), + (None, None), + ], +) +def test_publication_number(value: Any, expected: str | None) -> None: + """Test publication numbers reduce to digits inside wildcards.""" + assert query.publication_number("f", value) == expected + + +def test_either() -> None: + """Test either joins clauses with OR inside parentheses.""" + assert query.either("a:1", "b:2", None) == "(a:1 OR b:2)" + assert query.either("a:1", None) == "a:1" + assert query.either(None, None) is None + + +def test_all_of() -> None: + """Test all_of joins clauses with AND, skipping absent ones.""" + assert query.all_of("a:1", None, "b:2") == "a:1 AND b:2" + assert query.all_of(None) is None diff --git a/tests/integration/test_enriched_citations_integration.py b/tests/integration/test_enriched_citations_integration.py index af6f456..910c5e5 100644 --- a/tests/integration/test_enriched_citations_integration.py +++ b/tests/integration/test_enriched_citations_integration.py @@ -47,6 +47,32 @@ def test_search_by_application_number( for doc in response.docs: assert doc.patent_application_number == "15061308" + def test_search_by_series_code_application_number( + self, enriched_citations_client: EnrichedCitationsClient + ) -> None: + """Test a series-code patent_application_number_q (NN/NNNNNN) matches live.""" + response = enriched_citations_client.search_citations( + patent_application_number_q="15/061308", rows=50 + ) + + assert response.num_found == 25 + assert [doc.patent_application_number for doc in response.docs] == [ + "15061308" + ] * 25 + + def test_search_by_office_action_category_or_expression( + self, enriched_citations_client: EnrichedCitationsClient + ) -> None: + """Test an OR expression in office_action_category_q matches live.""" + response = enriched_citations_client.search_citations( + patent_application_number_q="15061308", + office_action_category_q="CTNF OR CTFR", + rows=50, + ) + + assert response.num_found == 25 + assert [doc.office_action_category for doc in response.docs] == ["CTNF"] * 25 + def test_search_by_citation_category_code( self, enriched_citations_client: EnrichedCitationsClient ) -> None: diff --git a/tests/integration/test_oa_actions_integration.py b/tests/integration/test_oa_actions_integration.py index b0b6114..f5771ef 100644 --- a/tests/integration/test_oa_actions_integration.py +++ b/tests/integration/test_oa_actions_integration.py @@ -57,6 +57,20 @@ def test_search_by_application_number( for doc in response.docs: assert _KNOWN_APP_NUMBER in doc.patent_application_number + def test_search_by_series_code_application_number( + self, oa_actions_client: OAActionsClient + ) -> None: + """Test a series-code patent_application_number_q (NN/NNNNNN) matches live.""" + response = oa_actions_client.search( + patent_application_number_q="11/363598", rows=10 + ) + + assert response.num_found == 2 + assert [doc.patent_application_number for doc in response.docs] == [ + ["11363598"], + ["11363598"], + ] + def test_search_by_id_exact_values( self, oa_actions_client: OAActionsClient ) -> None: diff --git a/tests/integration/test_oa_citations_integration.py b/tests/integration/test_oa_citations_integration.py index ff6e06e..766ffe0 100644 --- a/tests/integration/test_oa_citations_integration.py +++ b/tests/integration/test_oa_citations_integration.py @@ -58,6 +58,19 @@ def test_search_by_application_number( for doc in response.docs: assert doc.patent_application_number == _KNOWN_APP_NUMBER + def test_search_by_series_code_application_number( + self, oa_citations_client: OACitationsClient + ) -> None: + """Test a series-code patent_application_number_q (NN/NNNNNN) matches live.""" + response = oa_citations_client.search( + patent_application_number_q="17/519936", rows=20 + ) + + assert response.num_found == 15 + assert [doc.patent_application_number for doc in response.docs] == [ + "17519936" + ] * 15 + def test_search_by_id_exact_values( self, oa_citations_client: OACitationsClient ) -> None: diff --git a/tests/integration/test_oa_rejections_integration.py b/tests/integration/test_oa_rejections_integration.py index bc7b6d4..e0d9b05 100644 --- a/tests/integration/test_oa_rejections_integration.py +++ b/tests/integration/test_oa_rejections_integration.py @@ -57,6 +57,19 @@ def test_search_by_application_number( doc.patent_application_number == _KNOWN_APP_NUMBER for doc in response.docs ) + def test_search_by_series_code_application_number( + self, oa_rejections_client: OARejectionsClient + ) -> None: + """Test a series-code patent_application_number_q (NN/NNNNNN) matches live.""" + response = oa_rejections_client.search( + patent_application_number_q="12/190351", rows=50 + ) + + assert response.num_found == 31 + assert [doc.patent_application_number for doc in response.docs] == [ + "12190351" + ] * 31 + def test_search_by_doc_code(self, oa_rejections_client: OARejectionsClient) -> None: response = oa_rejections_client.search( legacy_document_code_identifier_q="CTNF", rows=5 diff --git a/tests/integration/test_patent_data_integration.py b/tests/integration/test_patent_data_integration.py index feb709c..544eb0f 100644 --- a/tests/integration/test_patent_data_integration.py +++ b/tests/integration/test_patent_data_integration.py @@ -136,6 +136,89 @@ def test_get_application_by_number( assert isinstance(patent_wrapper.application_meta_data, ApplicationMetaData) assert patent_wrapper.application_meta_data.invention_title is not None + def test_get_application_by_number_series_code_input( + self, + patent_data_client: PatentDataClient, + ) -> None: + """Test a series-code application number (NN/NNN,NNN) resolves live.""" + patent_wrapper = patent_data_client.get_application_by_number("14/412,875") + + assert patent_wrapper.application_number_text == "14412875" + + def test_search_applications_series_code_application_number_q( + self, + patent_data_client: PatentDataClient, + ) -> None: + """Test application_number_q with a series code (NN/NNNNNN) matches live.""" + response = patent_data_client.search_applications( + application_number_q="14/412875", limit=1 + ) + + assert response.count == 1 + assert [ + w.application_number_text for w in response.patent_file_wrapper_data_bag + ] == ["14412875"] + + def test_search_applications_or_expression_application_number_q( + self, + patent_data_client: PatentDataClient, + ) -> None: + """Test application_number_q applies the field to every term of an OR.""" + response = patent_data_client.search_applications( + application_number_q="14412875 OR 10000000", limit=10 + ) + + assert response.count == 1 + assert [ + w.application_number_text for w in response.patent_file_wrapper_data_bag + ] == ["14412875"] + + def test_search_applications_inventor_name_with_quote( + self, + patent_data_client: PatentDataClient, + ) -> None: + """Test a double quote inside inventor_name_q is escaped live.""" + response = patent_data_client.search_applications( + inventor_name_q='Smith "John', + filing_date_from_q="2000-01-01", + filing_date_to_q="2020-12-31", + limit=10, + ) + + assert response.count == 1 + assert [ + w.application_number_text for w in response.patent_file_wrapper_data_bag + ] == ["09890205"] + + def test_search_applications_patent_number_or_expression( + self, + patent_data_client: PatentDataClient, + ) -> None: + """Test patent_number_q applies the field to every term of an OR live.""" + response = patent_data_client.search_applications( + patent_number_q="10000000 OR 10000001", limit=10 + ) + + assert response.count == 2 + assert sorted( + w.application_meta_data.patent_number + for w in response.patent_file_wrapper_data_bag + ) == ["10000000", "10000001"] + + def test_search_applications_pct_application_number_q( + self, + patent_data_client: PatentDataClient, + ) -> None: + """Test application_number_q with a PCT number (PCT/CCYYYY/NNNNNN) matches live.""" + response = patent_data_client.search_applications( + application_number_q="PCT/US2025/047756", limit=1 + ) + + assert response.count == 1 + assert [ + w.application_number_text for w in response.patent_file_wrapper_data_bag + ] == ["PCTUS2025047756"] + def test_to_dict_matches_raw_api_response(self, api_key: str | None) -> None: """Test that to_dict() output matches the original API response stored in raw_data. diff --git a/tests/integration/test_petition_decisions_integration.py b/tests/integration/test_petition_decisions_integration.py index 057f9cf..c5ba743 100644 --- a/tests/integration/test_petition_decisions_integration.py +++ b/tests/integration/test_petition_decisions_integration.py @@ -195,6 +195,19 @@ def test_search_decisions_with_application_number( except USPTOApiError as e: pytest.fail(f"Application number search failed: {e}") + def test_search_decisions_with_series_code_application_number( + self, petition_decisions_client: PetitionDecisionsClient + ) -> None: + """Test a series-code application_number_q (NN/NNNNNN) matches live.""" + response = petition_decisions_client.search_decisions( + application_number_q="17/765301", limit=10 + ) + + assert response.count == 1 + assert [ + d.application_number_text for d in response.petition_decision_data_bag + ] == ["17765301"] + def test_search_decisions_with_patent_number( self, petition_decisions_client: PetitionDecisionsClient ) -> None: diff --git a/tests/integration/test_ptab_appeals_integration.py b/tests/integration/test_ptab_appeals_integration.py index 309a766..15beba8 100644 --- a/tests/integration/test_ptab_appeals_integration.py +++ b/tests/integration/test_ptab_appeals_integration.py @@ -12,7 +12,7 @@ from pyUSPTO.clients import PTABAppealsClient from pyUSPTO.config import USPTOConfig -from pyUSPTO.exceptions import USPTOApiError +from pyUSPTO.exceptions import USPTOApiNotFoundError from pyUSPTO.models.ptab import AppealMetaData, PTABAppealDecision, PTABAppealResponse # Import shared fixtures @@ -66,57 +66,77 @@ class TestPTABAppealsIntegration: def test_search_decisions_get(self, ptab_appeals_client: PTABAppealsClient) -> None: """Test searching PTAB appeal decisions using GET method.""" - try: - response = ptab_appeals_client.search_decisions( - query="appealMetaData.applicationTypeCategory:REGULAR", - limit=2, - ) - - assert response is not None - assert isinstance(response, PTABAppealResponse) - assert response.count >= 0 + response = ptab_appeals_client.search_decisions( + query="appealMetaData.applicationTypeCategory:REGULAR", + limit=2, + ) - if response.count > 0: - assert response.patent_appeal_data_bag is not None - assert len(response.patent_appeal_data_bag) > 0 - assert len(response.patent_appeal_data_bag) <= 2 + assert response is not None + assert isinstance(response, PTABAppealResponse) + assert response.count >= 0 - decision = response.patent_appeal_data_bag[0] - assert isinstance(decision, PTABAppealDecision) - assert decision.appeal_number is not None - else: - pytest.fail("There should always be a response to this query.") + if response.count > 0: + assert response.patent_appeal_data_bag is not None + assert len(response.patent_appeal_data_bag) > 0 + assert len(response.patent_appeal_data_bag) <= 2 - except USPTOApiError as e: - pytest.fail(f"PTAB Appeals API error during search_decisions GET: {e}") + decision = response.patent_appeal_data_bag[0] + assert isinstance(decision, PTABAppealDecision) + assert decision.appeal_number is not None + else: + pytest.fail("There should always be a response to this query.") def test_search_decisions_with_convenience_params( self, ptab_appeals_client: PTABAppealsClient ) -> None: """Test searching appeal decisions with application number.""" - try: - # Use direct query since applicationNumberText is nested under appellantData - response = ptab_appeals_client.search_decisions( - query="appellantData.applicationNumberText:12608694", - limit=2, - ) + # Use direct query since applicationNumberText is nested under appellantData + response = ptab_appeals_client.search_decisions( + query="appellantData.applicationNumberText:12608694", + limit=2, + ) + + assert response is not None + assert isinstance(response, PTABAppealResponse) + assert response.count >= 0 + + if response.count > 0: + assert response.patent_appeal_data_bag is not None + for decision in response.patent_appeal_data_bag: + assert isinstance(decision, PTABAppealDecision) + # Verify application type if metadata present + if decision.appeal_meta_data: + assert isinstance(decision.appeal_meta_data, AppealMetaData) - assert response is not None - assert isinstance(response, PTABAppealResponse) - assert response.count >= 0 + def test_search_decisions_multiword_appellant_name( + self, ptab_appeals_client: PTABAppealsClient + ) -> None: + """Test a multi-word appellant_name_q is searched as one phrase live.""" + response = ptab_appeals_client.search_decisions( + appellant_name_q="Block, Inc.", + decision_date_from_q="2025-01-01", + decision_date_to_q="2025-12-31", + limit=10, + ) + + assert response.count == 1 + assert [d.appeal_number for d in response.patent_appeal_data_bag] == [ + "2025003211" + ] + + def test_search_decisions_with_series_code_application_number( + self, ptab_appeals_client: PTABAppealsClient + ) -> None: + """Test a series-code application_number_text_q (NN/NNNNNN) matches live.""" + response = ptab_appeals_client.search_decisions( + application_number_text_q="18/136874", limit=10 + ) - if response.count > 0: - assert response.patent_appeal_data_bag is not None - for decision in response.patent_appeal_data_bag: - assert isinstance(decision, PTABAppealDecision) - # Verify application type if metadata present - if decision.appeal_meta_data: - assert isinstance(decision.appeal_meta_data, AppealMetaData) - - except USPTOApiError as e: - pytest.fail( - f"PTAB Appeals API error during search_decisions with convenience params: {e}" - ) + assert response.count == 1 + assert [ + d.appellant_data.application_number_text + for d in response.patent_appeal_data_bag + ] == ["18136874"] def test_search_decisions_post( self, ptab_appeals_client: PTABAppealsClient @@ -127,92 +147,73 @@ def test_search_decisions_post( "pagination": {"offset": 0, "limit": 2}, } - try: - response = ptab_appeals_client.search_decisions(post_body=post_body) - - assert response is not None - assert isinstance(response, PTABAppealResponse) - assert response.count >= 0 + response = ptab_appeals_client.search_decisions(post_body=post_body) - if response.count > 0: - assert response.patent_appeal_data_bag is not None - assert len(response.patent_appeal_data_bag) <= 2 + assert response is not None + assert isinstance(response, PTABAppealResponse) + assert response.count >= 0 - except USPTOApiError as e: - pytest.fail(f"PTAB Appeals API error during search_decisions POST: {e}") + if response.count > 0: + assert response.patent_appeal_data_bag is not None + assert len(response.patent_appeal_data_bag) <= 2 def test_search_decisions_with_date_filters( self, ptab_appeals_client: PTABAppealsClient ) -> None: """Test searching appeal decisions with date range filters.""" - try: - # Use direct query with correct field name (decisionIssueDate, not decisionDate) - response = ptab_appeals_client.search_decisions( - query="decisionData.decisionIssueDate:[2014-01-01 TO 2020-12-31]", - limit=2, - ) - - assert response is not None - assert isinstance(response, PTABAppealResponse) - assert response.count >= 0 + # Use direct query with correct field name (decisionIssueDate, not decisionDate) + response = ptab_appeals_client.search_decisions( + query="decisionData.decisionIssueDate:[2014-01-01 TO 2020-12-31]", + limit=2, + ) - if response.count > 0: - assert response.patent_appeal_data_bag is not None + assert response is not None + assert isinstance(response, PTABAppealResponse) + assert response.count >= 0 - except USPTOApiError as e: - pytest.fail( - f"PTAB Appeals API error during search_decisions with date filters: {e}" - ) + if response.count > 0: + assert response.patent_appeal_data_bag is not None def test_search_decisions_by_decision_type( self, ptab_appeals_client: PTABAppealsClient ) -> None: """Test searching appeal decisions by decision type.""" - try: - # Use direct query since decisionTypeCategory is nested under decisionData - response = ptab_appeals_client.search_decisions( - query="decisionData.decisionTypeCategory:Decision", - limit=2, - ) - - assert response is not None - assert isinstance(response, PTABAppealResponse) - assert response.count >= 0 - - if response.count > 0: - assert response.patent_appeal_data_bag is not None - for decision in response.patent_appeal_data_bag: - # Verify decision type if present - if decision.decision_data: - if decision.decision_data.decision_type_category: - assert ( - "Decision" - in decision.decision_data.decision_type_category - ) - - except USPTOApiError as e: - pytest.fail(f"PTAB Appeals API error during search by decision type: {e}") + # Use direct query since decisionTypeCategory is nested under decisionData + response = ptab_appeals_client.search_decisions( + query="decisionData.decisionTypeCategory:Decision", + limit=2, + ) + + assert response is not None + assert isinstance(response, PTABAppealResponse) + assert response.count >= 0 + + if response.count > 0: + assert response.patent_appeal_data_bag is not None + for decision in response.patent_appeal_data_bag: + # Verify decision type if present + if decision.decision_data: + if decision.decision_data.decision_type_category: + assert ( + "Decision" in decision.decision_data.decision_type_category + ) def test_search_decisions_by_appellant( self, ptab_appeals_client: PTABAppealsClient ) -> None: """Test searching appeal decisions by inventor name.""" - try: - # Search by inventor name (realPartyInInterestName contains inventor info) - response = ptab_appeals_client.search_decisions( - query="appellantData.inventorName:*", - limit=2, - ) + # Search by inventor name (realPartyInInterestName contains inventor info) + response = ptab_appeals_client.search_decisions( + query="appellantData.inventorName:*", + limit=2, + ) - assert response is not None - assert isinstance(response, PTABAppealResponse) - assert response.count >= 0 + assert response is not None + assert isinstance(response, PTABAppealResponse) + assert response.count >= 0 - if response.count > 0: - assert response.patent_appeal_data_bag is not None - - except USPTOApiError as e: - pytest.fail(f"PTAB Appeals API error during search by appellant: {e}") + if response.count > 0: + assert response.patent_appeal_data_bag is not None def test_paginate_decisions(self, ptab_appeals_client: PTABAppealsClient) -> None: """Test paginating through appeal decisions.""" @@ -221,44 +222,34 @@ def test_paginate_decisions(self, ptab_appeals_client: PTABAppealsClient) -> Non total_decisions = 0 - try: - for decision in ptab_appeals_client.paginate_decisions( - query="appealMetaData.applicationTypeCategory:REGULAR", - limit=page_size, - ): - assert isinstance(decision, PTABAppealDecision) - assert decision.appeal_number is not None - - total_decisions += 1 + for decision in ptab_appeals_client.paginate_decisions( + query="appealMetaData.applicationTypeCategory:REGULAR", + limit=page_size, + ): + assert isinstance(decision, PTABAppealDecision) + assert decision.appeal_number is not None - if total_decisions >= max_decisions: - break + total_decisions += 1 - assert total_decisions > 0, "Should have retrieved at least one decision" + if total_decisions >= max_decisions: + break - except USPTOApiError as e: - pytest.fail(f"Pagination test failed with API error: {e}") + assert total_decisions > 0, "Should have retrieved at least one decision" def test_search_with_optional_params( self, ptab_appeals_client: PTABAppealsClient ) -> None: """Test searching with optional parameters like sort and facets.""" - try: - response = ptab_appeals_client.search_decisions( - query="appealMetaData.applicationTypeCategory:REGULAR", - limit=2, - sort="appealNumber desc", - offset=0, - ) + response = ptab_appeals_client.search_decisions( + query="appealMetaData.applicationTypeCategory:REGULAR", + limit=2, + sort="appealNumber desc", + offset=0, + ) - assert response is not None - assert isinstance(response, PTABAppealResponse) - assert response.count >= 0 - - except USPTOApiError as e: - pytest.fail( - f"PTAB Appeals API error during search with optional params: {e}" - ) + assert response is not None + assert isinstance(response, PTABAppealResponse) + assert response.count >= 0 def test_to_dict_matches_raw_api_response(self, api_key: str | None) -> None: """Test that to_dict() output matches the original API response stored in raw_data. @@ -270,104 +261,94 @@ def test_to_dict_matches_raw_api_response(self, api_key: str | None) -> None: config_with_raw = USPTOConfig(api_key=api_key, include_raw_data=True) client_with_raw = PTABAppealsClient(config=config_with_raw) - try: - # Search for a known appeal number - response = client_with_raw.search_decisions( - query="appealNumber:2015000194", - limit=1, - ) + # Search for a known appeal number + response = client_with_raw.search_decisions( + query="appealNumber:2015000194", + limit=1, + ) - if response is None or response.count == 0: - pytest.fail("No decision found for raw API comparison test") + if response is None or response.count == 0: + pytest.fail("No decision found for raw API comparison test") - assert response.raw_data is not None, ( - "raw_data should be populated when include_raw_data=True" - ) + assert response.raw_data is not None, ( + "raw_data should be populated when include_raw_data=True" + ) - # PTAB models store raw_data as dict (not JSON string like other models) - assert isinstance(response.raw_data, dict), ( - "raw_data should be a dictionary" - ) + # PTAB models store raw_data as dict (not JSON string like other models) + assert isinstance(response.raw_data, dict), "raw_data should be a dictionary" - # Get the raw API response dict - raw_api_dict = response.raw_data + # Get the raw API response dict + raw_api_dict = response.raw_data - # Convert the model back to dict - to_dict_output = response.to_dict() + # Convert the model back to dict + to_dict_output = response.to_dict() - # Deep comparison function - def compare_dicts(dict1, dict2, path=""): - """Recursively compare two dictionaries and report differences.""" - differences = [] + # Deep comparison function + def compare_dicts(dict1, dict2, path=""): + """Recursively compare two dictionaries and report differences.""" + differences = [] - # Check keys present in dict1 but not dict2 - keys1 = set(dict1.keys()) - keys2 = set(dict2.keys()) + # Check keys present in dict1 but not dict2 + keys1 = set(dict1.keys()) + keys2 = set(dict2.keys()) - missing_in_dict2 = keys1 - keys2 - if missing_in_dict2: - differences.append( - f"Keys in to_dict but not in raw API at {path}: {missing_in_dict2}" - ) + missing_in_dict2 = keys1 - keys2 + if missing_in_dict2: + differences.append( + f"Keys in to_dict but not in raw API at {path}: {missing_in_dict2}" + ) - missing_in_dict1 = keys2 - keys1 - if missing_in_dict1: - differences.append( - f"Keys in raw API but not in to_dict at {path}: {missing_in_dict1}" - ) + missing_in_dict1 = keys2 - keys1 + if missing_in_dict1: + differences.append( + f"Keys in raw API but not in to_dict at {path}: {missing_in_dict1}" + ) - # Compare values for common keys - for key in keys1 & keys2: - val1 = dict1[key] - val2 = dict2[key] - current_path = f"{path}.{key}" if path else key + # Compare values for common keys + for key in keys1 & keys2: + val1 = dict1[key] + val2 = dict2[key] + current_path = f"{path}.{key}" if path else key - if type(val1) is not type(val2): - differences.append( - f"Type mismatch at {current_path}: {type(val1).__name__} vs {type(val2).__name__}" - ) - elif isinstance(val1, dict): - differences.extend(compare_dicts(val1, val2, current_path)) - elif isinstance(val1, list): - if len(val1) != len(val2): - differences.append( - f"List length mismatch at {current_path}: {len(val1)} vs {len(val2)}" - ) - else: - for i, (item1, item2) in enumerate(zip(val1, val2)): - if isinstance(item1, dict) and isinstance(item2, dict): - differences.extend( - compare_dicts( - item1, item2, f"{current_path}[{i}]" - ) - ) - elif item1 != item2: - differences.append( - f"Value mismatch at {current_path}[{i}]: {item1!r} vs {item2!r}" - ) - elif val1 != val2: + if type(val1) is not type(val2): + differences.append( + f"Type mismatch at {current_path}: {type(val1).__name__} vs {type(val2).__name__}" + ) + elif isinstance(val1, dict): + differences.extend(compare_dicts(val1, val2, current_path)) + elif isinstance(val1, list): + if len(val1) != len(val2): differences.append( - f"Value mismatch at {current_path}: {val1!r} vs {val2!r}" + f"List length mismatch at {current_path}: {len(val1)} vs {len(val2)}" ) + else: + for i, (item1, item2) in enumerate(zip(val1, val2)): + if isinstance(item1, dict) and isinstance(item2, dict): + differences.extend( + compare_dicts(item1, item2, f"{current_path}[{i}]") + ) + elif item1 != item2: + differences.append( + f"Value mismatch at {current_path}[{i}]: {item1!r} vs {item2!r}" + ) + elif val1 != val2: + differences.append( + f"Value mismatch at {current_path}: {val1!r} vs {val2!r}" + ) - return differences - - # Perform the comparison - differences = compare_dicts(to_dict_output, raw_api_dict) + return differences - # If there are differences, print them and fail - if differences: - diff_report = "\n".join( - differences[:20] - ) # Limit to first 20 differences - if len(differences) > 20: - diff_report += f"\n... and {len(differences) - 20} more differences" - pytest.fail( - f"to_dict() output does not match raw API response. Differences found:\n{diff_report}" - ) + # Perform the comparison + differences = compare_dicts(to_dict_output, raw_api_dict) - except USPTOApiError as e: - pytest.fail(f"Raw API comparison test failed with API error: {e}") + # If there are differences, print them and fail + if differences: + diff_report = "\n".join(differences[:20]) # Limit to first 20 differences + if len(differences) > 20: + diff_report += f"\n... and {len(differences) - 20} more differences" + pytest.fail( + f"to_dict() output does not match raw API response. Differences found:\n{diff_report}" + ) def test_download_appeal_archive( self, @@ -375,51 +356,32 @@ def test_download_appeal_archive( appeals_with_download_uris: PTABAppealResponse, ) -> None: """Test downloading appeal archive file without extraction.""" - try: + if ( + not appeals_with_download_uris + or not appeals_with_download_uris.patent_appeal_data_bag + ): + pytest.fail("No appeal decisions found for archive download test") + + for decision in appeals_with_download_uris.patent_appeal_data_bag: if ( - not appeals_with_download_uris - or not appeals_with_download_uris.patent_appeal_data_bag + not decision.appeal_meta_data + or not decision.appeal_meta_data.file_download_uri ): - pytest.fail("No appeal decisions found for archive download test") - - # Try multiple appeals until one downloads successfully - last_error = None - for decision in appeals_with_download_uris.patent_appeal_data_bag: - if ( - not decision.appeal_meta_data - or not decision.appeal_meta_data.file_download_uri - ): - continue - - try: - file_path = ptab_appeals_client.download_appeal_archive( - decision.appeal_meta_data, - destination=TEST_DOWNLOAD_DIR, - overwrite=True, - ) + continue - # If we get here, download succeeded - assert file_path is not None - assert os.path.exists(file_path) - assert os.path.getsize(file_path) > 0 - assert file_path.endswith((".zip", ".tar", ".tar.gz", ".tgz")) - return # Test passed! - - except USPTOApiError as e: - # Save error and try next appeal - last_error = e - continue - - # If we tried all appeals and none worked, fail with the last error - if last_error: - pytest.fail( - f"All appeal archives failed to download. Last error: {last_error}" - ) - else: - pytest.fail("No appeal with valid file_download_uri found") + file_path = ptab_appeals_client.download_appeal_archive( + decision.appeal_meta_data, + destination=TEST_DOWNLOAD_DIR, + overwrite=True, + ) - except USPTOApiError as e: - pytest.fail(f"PTAB Appeals API error during search: {e}") + assert file_path is not None + assert os.path.exists(file_path) + assert os.path.getsize(file_path) > 0 + assert file_path.endswith((".zip", ".tar", ".tar.gz", ".tgz")) + return + + pytest.fail("No appeal with valid file_download_uri found") def test_download_appeal_documents( self, @@ -427,96 +389,60 @@ def test_download_appeal_documents( appeals_with_download_uris: PTABAppealResponse, ) -> None: """Test downloading and extracting appeal documents.""" - try: + if ( + not appeals_with_download_uris + or not appeals_with_download_uris.patent_appeal_data_bag + ): + pytest.fail("No appeal decisions found for documents download test") + + for decision in appeals_with_download_uris.patent_appeal_data_bag: if ( - not appeals_with_download_uris - or not appeals_with_download_uris.patent_appeal_data_bag + not decision.appeal_meta_data + or not decision.appeal_meta_data.file_download_uri ): - pytest.fail("No appeal decisions found for documents download test") - - # Try multiple appeals until one downloads successfully - last_error = None - for decision in appeals_with_download_uris.patent_appeal_data_bag: - if ( - not decision.appeal_meta_data - or not decision.appeal_meta_data.file_download_uri - ): - continue - - try: - extracted_path = ptab_appeals_client.download_appeal_documents( - decision.appeal_meta_data, - destination=TEST_DOWNLOAD_DIR, - overwrite=True, - ) - - # If we get here, download and extraction succeeded - assert extracted_path is not None - assert os.path.exists(extracted_path) - return # Test passed! + continue - except USPTOApiError as e: - # Save error and try next appeal - last_error = e - continue + extracted_path = ptab_appeals_client.download_appeal_documents( + decision.appeal_meta_data, + destination=TEST_DOWNLOAD_DIR, + overwrite=True, + ) - # If we tried all appeals and none worked, fail with the last error - if last_error: - pytest.fail( - f"All appeal documents failed to download. Last error: {last_error}" - ) - else: - pytest.fail("No appeal with valid file_download_uri found") + assert extracted_path is not None + assert os.path.exists(extracted_path) + return - except USPTOApiError as e: - pytest.fail(f"PTAB Appeals API error during search: {e}") + pytest.fail("No appeal with valid file_download_uri found") def test_download_appeal_document( self, ptab_appeals_client: PTABAppealsClient ) -> None: """Test downloading individual appeal document.""" - try: - response = ptab_appeals_client.search_decisions( - query="appealMetaData.applicationTypeCategory:REGULAR", - limit=1, - ) - - if not response or not response.patent_appeal_data_bag: - pytest.fail("No appeal decisions found for document download test") + response = ptab_appeals_client.search_decisions( + query="appealMetaData.applicationTypeCategory:REGULAR", + limit=1, + ) - decision = response.patent_appeal_data_bag[0] - if ( - not decision.document_data - or not decision.document_data.file_download_uri - ): - pytest.fail("No file_download_uri available for test") + if not response or not response.patent_appeal_data_bag: + pytest.fail("No appeal decisions found for document download test") - file_path = ptab_appeals_client.download_appeal_document( - decision.document_data, - destination=TEST_DOWNLOAD_DIR, - overwrite=True, - ) + decision = response.patent_appeal_data_bag[0] + if not decision.document_data or not decision.document_data.file_download_uri: + pytest.fail("No file_download_uri available for test") - assert file_path is not None - assert os.path.exists(file_path) - assert os.path.getsize(file_path) > 0 + file_path = ptab_appeals_client.download_appeal_document( + decision.document_data, + destination=TEST_DOWNLOAD_DIR, + overwrite=True, + ) - except USPTOApiError as e: - pytest.fail(f"PTAB Appeals API error during document download: {e}") + assert file_path is not None + assert os.path.exists(file_path) + assert os.path.getsize(file_path) > 0 def test_invalid_query_handling( self, ptab_appeals_client: PTABAppealsClient ) -> None: - """Test proper error handling with an invalid query.""" - try: - # Use an obviously malformed query - response = ptab_appeals_client.search_decisions( - query="INVALID_FIELD:value", limit=1 - ) - - # API may return 0 results instead of error for invalid field - assert isinstance(response, PTABAppealResponse) - - except USPTOApiError as e: - # This is acceptable - API may return error for invalid queries - assert e.status_code in [400, 404, 500] + """Test a query on a field that does not exist raises a 404.""" + with pytest.raises(USPTOApiNotFoundError): + ptab_appeals_client.search_decisions(query="INVALID_FIELD:value", limit=1) diff --git a/tests/integration/test_ptab_interferences_integration.py b/tests/integration/test_ptab_interferences_integration.py index 2b98b4d..7bf7cb5 100644 --- a/tests/integration/test_ptab_interferences_integration.py +++ b/tests/integration/test_ptab_interferences_integration.py @@ -112,6 +112,20 @@ def test_search_decisions_with_convenience_params( f"PTAB Interferences API error during search_decisions with convenience params: {e}" ) + def test_search_decisions_with_series_code_application_number( + self, ptab_interferences_client: PTABInterferencesClient + ) -> None: + """Test a series-code senior party application number matches live.""" + response = ptab_interferences_client.search_decisions( + senior_party_application_number_q="14/443829", limit=10 + ) + + assert response.count == 2 + assert [ + d.senior_party_data.application_number_text + for d in response.patent_interference_data_bag + ] == ["14443829", "14443829"] + def test_search_decisions_post( self, ptab_interferences_client: PTABInterferencesClient ) -> None: diff --git a/tests/integration/test_ptab_trials_integration.py b/tests/integration/test_ptab_trials_integration.py index ab95517..1eea47b 100644 --- a/tests/integration/test_ptab_trials_integration.py +++ b/tests/integration/test_ptab_trials_integration.py @@ -133,6 +133,20 @@ def test_search_proceedings_post( except USPTOApiError as e: pytest.fail(f"PTAB Trials API error during search_proceedings POST: {e}") + def test_search_decisions_with_series_code_application_number( + self, ptab_trials_client: PTABTrialsClient + ) -> None: + """Test a series-code application_number_q (NN/NNNNNN) matches live.""" + response = ptab_trials_client.search_decisions( + application_number_q="15/217538", limit=10 + ) + + assert response.count == 4 + assert [ + d.patent_owner_data.application_number_text + for d in response.patent_trial_document_data_bag + ] == ["15217538"] * 4 + def test_search_documents_get(self, ptab_trials_client: PTABTrialsClient) -> None: """Test searching PTAB trial documents using GET method.""" try: diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py index f1ca418..a6341b1 100644 --- a/tests/test_exceptions.py +++ b/tests/test_exceptions.py @@ -115,6 +115,48 @@ def test_from_http_error_json_body_missing_fields(self) -> None: ) # Fallback to full text if JSON but no specific detail fields assert args.request_identifier is None + def test_from_http_error_403_missing_auth_token_adds_hint(self) -> None: + """Test a 403 'Missing Authentication Token' body adds the bad-path hint.""" + mock_http_error = MagicMock(spec=requests.exceptions.HTTPError) + mock_http_error.response = MagicMock(spec=requests.Response) + mock_http_error.response.status_code = 403 + mock_http_error.response.reason = "Forbidden" + mock_http_error.response.headers = {"content-type": "application/json"} + mock_http_error.response.json.return_value = { + "message": "Missing Authentication Token" + } + mock_http_error.response.text = '{"message":"Missing Authentication Token"}' + + args = APIErrorArgs.from_http_error( + http_error=mock_http_error, client_operation_message="Client op failed" + ) + assert args.message == ( + "Client op failed (USPTO also returns 'Missing Authentication Token' " + "when the requested URL path does not exist)" + ) + assert args.api_short_error == "Forbidden" + assert args.error_details == '{"message":"Missing Authentication Token"}' + + def test_from_http_error_403_other_body_keeps_message(self) -> None: + """Test a 403 with a normal USPTO error body keeps the original message.""" + mock_http_error = MagicMock(spec=requests.exceptions.HTTPError) + mock_http_error.response = MagicMock(spec=requests.Response) + mock_http_error.response.status_code = 403 + mock_http_error.response.reason = "Forbidden" + mock_http_error.response.headers = {"content-type": "application/json"} + mock_http_error.response.json.return_value = { + "error": "Forbidden", + "errorDetails": "Invalid API key", + } + mock_http_error.response.text = ( + '{"error": "Forbidden", "errorDetails": "Invalid API key"}' + ) + + args = APIErrorArgs.from_http_error( + http_error=mock_http_error, client_operation_message="Client op failed" + ) + assert args.message == "Client op failed" + @pytest.mark.parametrize( "body", [["unexpected", "list"], "bare string", 42], ids=["list", "str", "int"] )