"""config - Configuration management for USPTO API clients.
This module provides configuration management for USPTO API clients,
including API keys, base URLs, and HTTP transport settings.
"""
import os
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
import requests
from pyUSPTO.http_config import HTTPConfig
DEFAULT_BASE_URL = "https://api.uspto.gov"
def _env_values() -> dict[str, Any]:
"""Build a dict of config values from environment variables (or defaults)."""
return {
"api_key": os.environ.get("USPTO_API_KEY") or None,
"bulk_data_base_url": os.environ.get(
"USPTO_BULK_DATA_BASE_URL", DEFAULT_BASE_URL
),
"patent_data_base_url": os.environ.get(
"USPTO_PATENT_DATA_BASE_URL", DEFAULT_BASE_URL
),
"petition_decisions_base_url": os.environ.get(
"USPTO_PETITION_DECISIONS_BASE_URL", DEFAULT_BASE_URL
),
"ptab_base_url": os.environ.get("USPTO_PTAB_BASE_URL", DEFAULT_BASE_URL),
"enriched_citations_base_url": os.environ.get(
"USPTO_ENRICHED_CITATIONS_BASE_URL", DEFAULT_BASE_URL
),
"oa_actions_base_url": os.environ.get(
"USPTO_OA_ACTIONS_BASE_URL", DEFAULT_BASE_URL
),
"oa_rejections_base_url": os.environ.get(
"USPTO_OA_REJECTIONS_BASE_URL", DEFAULT_BASE_URL
),
"oa_citations_base_url": os.environ.get(
"USPTO_OA_CITATIONS_BASE_URL", DEFAULT_BASE_URL
),
"http_config": HTTPConfig.from_env(),
}
[docs]
class USPTOConfig:
"""Configuration for USPTO API clients.
Manages API-level configuration (keys, URLs) and optionally
accepts HTTP transport configuration via HTTPConfig.
"""
[docs]
def __init__(
self,
api_key: str | None = None,
bulk_data_base_url: str | None = None,
patent_data_base_url: str | None = None,
petition_decisions_base_url: str | None = None,
ptab_base_url: str | None = None,
enriched_citations_base_url: str | None = None,
oa_actions_base_url: str | None = None,
oa_rejections_base_url: str | None = None,
oa_citations_base_url: str | None = None,
http_config: HTTPConfig | None = None,
include_raw_data: bool = False,
):
"""Initialize the USPTOConfig.
Each field falls back to its environment variable (or the package
default) when the caller does not pass a value.
Args:
api_key: API key for authentication, defaults to USPTO_API_KEY environment variable
bulk_data_base_url: Base URL for the Bulk Data API
patent_data_base_url: Base URL for the Patent Data API
petition_decisions_base_url: Base URL for the Final Petition Decisions API
ptab_base_url: Base URL for the PTAB (Patent Trial and Appeal Board) API
enriched_citations_base_url: Base URL for the Enriched Citations API
oa_actions_base_url: Base URL for the Office Action Text Retrieval API
oa_rejections_base_url: Base URL for the Office Action Rejections API
oa_citations_base_url: Base URL for the Office Action Citations API
http_config: Optional HTTPConfig for request handling (uses defaults if None)
include_raw_data: If True, store raw JSON in response objects for debugging (default: False)
"""
env = _env_values()
self.api_key = api_key if api_key else env["api_key"]
self.bulk_data_base_url = (
bulk_data_base_url if bulk_data_base_url else env["bulk_data_base_url"]
)
self.patent_data_base_url = (
patent_data_base_url
if patent_data_base_url
else env["patent_data_base_url"]
)
self.petition_decisions_base_url = (
petition_decisions_base_url
if petition_decisions_base_url
else env["petition_decisions_base_url"]
)
self.ptab_base_url = ptab_base_url if ptab_base_url else env["ptab_base_url"]
self.enriched_citations_base_url = (
enriched_citations_base_url
if enriched_citations_base_url
else env["enriched_citations_base_url"]
)
self.oa_actions_base_url = (
oa_actions_base_url if oa_actions_base_url else env["oa_actions_base_url"]
)
self.oa_rejections_base_url = (
oa_rejections_base_url
if oa_rejections_base_url
else env["oa_rejections_base_url"]
)
self.oa_citations_base_url = (
oa_citations_base_url
if oa_citations_base_url
else env["oa_citations_base_url"]
)
self.http_config = http_config if http_config else env["http_config"]
# Control whether to include raw JSON data in response objects
self.include_raw_data = include_raw_data
# Session for all clients using this config (created lazily)
self._session: requests.Session | None = None
[docs]
@classmethod
def from_env(cls) -> "USPTOConfig":
"""Create a USPTOConfig from environment variables.
Returns:
USPTOConfig instance with values from environment
"""
return cls(**_env_values())
@property
def session(self) -> "requests.Session":
"""Get the HTTP session for this config, creating it if needed.
The session is created lazily on first access and reused for all
subsequent requests. All clients sharing this config will use the
same session for connection pooling.
Returns:
Session: The requests Session object with configured adapters.
"""
if self._session is None:
self._session = self._create_session()
return self._session
def _create_session(self) -> "requests.Session":
"""Create and configure a new requests Session.
Returns:
Session: Configured session with retry logic and connection pooling.
"""
from requests import Session
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = Session()
# Set API key header
if self.api_key:
session.headers["X-API-KEY"] = self.api_key
session.headers["Accept"] = "application/json"
# Apply custom headers from HTTP config
if self.http_config.custom_headers:
session.headers.update(self.http_config.custom_headers)
# Configure retry strategy
retry_strategy = Retry(
total=self.http_config.max_retries,
backoff_factor=self.http_config.backoff_factor,
status_forcelist=self.http_config.retry_status_codes,
allowed_methods={"GET", "POST"},
)
# Configure connection pooling
adapter = HTTPAdapter(
max_retries=retry_strategy,
pool_connections=self.http_config.pool_connections,
pool_maxsize=self.http_config.pool_maxsize,
)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
[docs]
def close(self) -> None:
"""Close the HTTP session and release resources.
This should be called when you're done using this config and all
clients created from it. After calling close(), the session will
be recreated if accessed again.
Example:
config = USPTOConfig(api_key="...")
client = PatentDataClient(config=config)
try:
# Use client
pass
finally:
config.close()
"""
if self._session is not None:
self._session.close()
self._session = None
[docs]
def __enter__(self) -> "USPTOConfig":
"""Enter context manager."""
return self
[docs]
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: object | None,
) -> None:
"""Exit context manager, closing the session."""
self.close()