From 680b5344dbf64e275bf13c04dfbbe4925081c900 Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 13:26:39 -0400 Subject: [PATCH 01/27] Enums for job search query params --- usajobsapi/endpoints/search.py | 61 ++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 7159f3c..6830c58 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -1,13 +1,74 @@ """Wrapper for the Job Search API.""" +from enum import StrEnum from typing import Dict from pydantic import BaseModel from usajobsapi.utils import _dump_by_alias +# Enums for query-params +# --- + +class SortField(StrEnum): + OPEN_DATE = "opendate" + CLOSED_DATE = "closedate" + ORGANIZATION_NAME = "organizationname" + JOB_TITLE = "jobtitle" + POSITION_TITLE = "positiontitle" + OPENING_DATE = "openingdate" + CLOSING_DATE = "closingdate" + HO_NAME = "honame" + SALARY_MIN = "salarymin" + LOCATION = "location" + DEPARTMENT = "department" + TITLE = "title" + AGENCY = "agency" + SALARY = "salary" + + +class SortDirection(StrEnum): + ASC = "Asc" + DESC = "Desc" + + +class WhoMayApply(StrEnum): + ALL = "All" + PUBLIC = "Public" + STATUS = "Status" + + +class Fields(StrEnum): + MIN = "Min" + FULL = "Full" + + +class HiringPath(StrEnum): + PUBLIC = "public" + VET = "vet" + N_GUARD = "nguard" + DISABILITY = "disability" + NATIVE = "native" + M_SPOUSE = "mspouse" + STUDENT = "student" + SES = "ses" + PEACE = "peace" + OVERSEAS = "overseas" + FED_INTERNAL_SEARCH = "fed-internal-search" + GRADUATES = "graduates" + FED_EXCEPTED = "fed-excepted" + FED_COMPETITIVE = "fed-competitive" + FED_TRANSITION = "fed-transition" + LAND = "land" + SPECIAL_AUTHORITIES = "special-authorities" + + +# Endpoint declaration +# --- class SearchEndpoint(BaseModel): + """Declarative endpoint definition for the Job Search API.""" + method: str = "GET" path: str = "/api/search" From 33bff5d24103c9e2664ba61c76c3423279a0c63e Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 13:28:33 -0400 Subject: [PATCH 02/27] JobSummary resp shape for SearchEndpoint --- usajobsapi/endpoints/search.py | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 6830c58..9834066 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -1,9 +1,9 @@ """Wrapper for the Job Search API.""" from enum import StrEnum -from typing import Dict +from typing import Dict, Optional -from pydantic import BaseModel +from pydantic import BaseModel, Field from usajobsapi.utils import _dump_by_alias @@ -73,8 +73,25 @@ class SearchEndpoint(BaseModel): path: str = "/api/search" class Params(BaseModel): + """Query-params""" + def to_params(self) -> Dict[str, str]: return _dump_by_alias(self) + # Response shapes + # --- + class JobSummary(BaseModel): + id: str = Field(alias="MatchedObjectId") + position_title: str = Field(alias="PositionTitle") + organization_name: Optional[str] = Field(default=None, alias="OrganizationName") + locations_display: Optional[str] = Field( + default=None, alias="PositionLocationDisplay" + ) + min_salary: Optional[float] = Field(default=None, alias="MinimumRange") + max_salary: Optional[float] = Field(default=None, alias="MaximumRange") + application_close_date: Optional[str] = Field( + default=None, alias="ApplicationCloseDate" + ) + class Response(BaseModel): pass From 7a3baa355e4cc598992d3e7da67fdad8fde0c714 Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 13:39:00 -0400 Subject: [PATCH 03/27] Add query param fields to SearchEndpoint.Params --- usajobsapi/endpoints/search.py | 108 +++++++++++++++++++++++++++++++-- 1 file changed, 104 insertions(+), 4 deletions(-) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 9834066..c279ec3 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -1,9 +1,9 @@ """Wrapper for the Job Search API.""" from enum import StrEnum -from typing import Dict, Optional +from typing import Annotated, Dict, List, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field from usajobsapi.utils import _dump_by_alias @@ -67,13 +67,113 @@ class HiringPath(StrEnum): # Endpoint declaration # --- class SearchEndpoint(BaseModel): - """Declarative endpoint definition for the Job Search API.""" + """ + Declarative endpoint definition for the Job Search API. + + Includes the endpoint's: + + - Parameters + - Response shapes + - Metadata + """ method: str = "GET" path: str = "/api/search" class Params(BaseModel): - """Query-params""" + """Declarative query-parameter model""" + + model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) + + keyword: Optional[str] = Field(None, serialization_alias="Keyword") + position_title: Optional[str] = Field(None, serialization_alias="PositionTitle") + + remuneration_min: Optional[int] = Field( + None, serialization_alias="RemunerationMinimumAmount" + ) + remuneration_max: Optional[int] = Field( + None, serialization_alias="RemunerationMaximumAmount" + ) + pay_grade_high: Optional[str] = Field(None, serialization_alias="PayGradeHigh") + pay_grade_low: Optional[str] = Field(None, serialization_alias="PayGradeLow") + + job_category_codes: List[str] = Field( + default_factory=list, serialization_alias="JobCategoryCode" + ) + position_schedule_type_codes: List[str] = Field( + default_factory=list, serialization_alias="PositionScheduleTypeCode" + ) + position_offering_type_codes: List[str] = Field( + default_factory=list, serialization_alias="PositionOfferingTypeCode" + ) + + organization: List[str] = Field( + default_factory=list, serialization_alias="Organization" + ) + location_names: List[str] = Field( + default_factory=list, serialization_alias="LocationName" + ) + radius: Annotated[ + Optional[int], Field(serialization_alias="Radius", strict=True, gt=0) + ] = None + + travel_percentage: List[str] = Field( + default_factory=list, serialization_alias="TravelPercentage" + ) + relocation: Optional[bool] = Field( + None, serialization_alias="RelocationIndicator" + ) + security_clearance_required: List[str] = Field( + default_factory=list, serialization_alias="SecurityClearanceRequired" + ) + position_sensitivity: List[str] = Field( + default_factory=list, serialization_alias="PositionSensitivity" + ) + + who_may_apply: Optional[WhoMayApply] = Field( + None, serialization_alias="WhoMayApply" + ) + hiring_paths: List[HiringPath] = Field( + default_factory=list, serialization_alias="HiringPath" + ) + + salary_bucket: List[str] = Field( + default_factory=list, serialization_alias="SalaryBucket" + ) + grade_bucket: List[str] = Field( + default_factory=list, serialization_alias="GradeBucket" + ) + + supervisory_status: Optional[str] = Field( + None, serialization_alias="SupervisoryStatus" + ) + date_posted_days: Annotated[ + Optional[int], + Field(serialization_alias="DatePosted", strict=True, ge=0, le=60), + ] = None + job_grade_codes: List[str] = Field( + default_factory=list, serialization_alias="JobGradeCode" + ) + mission_critical_tags: List[str] = Field( + default_factory=list, serialization_alias="MissionCriticalTags" + ) + + sort_field: Optional[SortField] = Field(None, serialization_alias="SortField") + sort_direction: Optional[SortDirection] = Field( + None, serialization_alias="SortDirection" + ) + page: Annotated[ + Optional[int], Field(serialization_alias="Page", strict=True, ge=1) + ] = None + results_per_page: Annotated[ + Optional[int], + Field(serialization_alias="ResultsPerPage", strict=True, ge=1, le=500), + ] = None + fields: Optional[Fields] = Field(None, serialization_alias="Fields") + + remote_indicator: Optional[bool] = Field( + None, serialization_alias="RemoteIndicator" + ) def to_params(self) -> Dict[str, str]: return _dump_by_alias(self) From 6601dd237d4da1c5b5831508f147dee18e93d2f4 Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 13:42:27 -0400 Subject: [PATCH 04/27] Some initial field validation for SearchEndpoint.Params --- usajobsapi/endpoints/search.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index c279ec3..1756dd5 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -3,7 +3,7 @@ from enum import StrEnum from typing import Annotated, Dict, List, Optional -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from usajobsapi.utils import _dump_by_alias @@ -175,6 +175,22 @@ class Params(BaseModel): None, serialization_alias="RemoteIndicator" ) + @model_validator(mode="after") + def _radius_requires_location(self) -> "SearchEndpoint.Params": + if self.radius is not None and not self.location_names: + raise ValueError("Radius requires at least one LocationName.") + return self + + @field_validator("remuneration_max") + @classmethod + def _check_min_le_max(cls, v, info): + mn = info.data.get("remuneration_min") + if v is not None and mn is not None and v < mn: + raise ValueError( + "RemunerationMaximumAmount must be >= RemunerationMinimumAmount." + ) + return v + def to_params(self) -> Dict[str, str]: return _dump_by_alias(self) From 67dd614d5a0cf06b79f60ddca1be50451c09fd27 Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 14:05:54 -0400 Subject: [PATCH 05/27] Remaining shapes for SearchEndpoint.Response --- usajobsapi/endpoints/search.py | 46 +++++++++++++++++++++++++++++++--- 1 file changed, 43 insertions(+), 3 deletions(-) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 1756dd5..fdb2f5e 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -1,9 +1,15 @@ """Wrapper for the Job Search API.""" from enum import StrEnum -from typing import Annotated, Dict, List, Optional +from typing import Annotated, Any, Dict, List, Optional -from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from pydantic import ( + BaseModel, + ConfigDict, + Field, + field_validator, + model_validator, +) from usajobsapi.utils import _dump_by_alias @@ -196,6 +202,7 @@ def to_params(self) -> Dict[str, str]: # Response shapes # --- + class JobSummary(BaseModel): id: str = Field(alias="MatchedObjectId") position_title: str = Field(alias="PositionTitle") @@ -209,5 +216,38 @@ class JobSummary(BaseModel): default=None, alias="ApplicationCloseDate" ) + class SearchResult(BaseModel): + result_count: Optional[int] = Field( + default=None, + alias="SearchResultCount", + ) + result_total: Optional[int] = Field( + default=None, + alias="SearchResultCountAll", + ) + items: List[Dict[str, Any]] = Field( + default_factory=list, + alias="SearchResultItems", + ) + + def jobs(self) -> List["SearchEndpoint.JobSummary"]: + # Convert to a normalized list versus keeping raw + out: List[SearchEndpoint.JobSummary] = [] + for item in self.items: + # Some responses nest the item under 'MatchedObjectDescriptor' + descriptor = item.get("MatchedObjectDescriptor") or item + try: + out.append(SearchEndpoint.JobSummary.model_validate(descriptor)) + except Exception: + continue + return out + class Response(BaseModel): - pass + language: Optional[str] = Field(default=None, alias="LanguageCode") + params: Optional["SearchEndpoint.Params"] = Field( + default=None, alias="SearchParameters" + ) + # Results are wrapped under SearchResult + search_result: Optional["SearchEndpoint.SearchResult"] = Field( + default=None, alias="SearchResult" + ) From 85a9fee677621cf307100e505b99fe9f46d3485f Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 14:09:37 -0400 Subject: [PATCH 06/27] Create placeholder test for SearchEndpoint --- tests/unit/test_search.py | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 tests/unit/test_search.py diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py new file mode 100644 index 0000000..877c54c --- /dev/null +++ b/tests/unit/test_search.py @@ -0,0 +1,6 @@ +"""Placeholder tests for USAJobs API package.""" + + +def test_placeholder() -> None: + """A trivial test to ensure the test suite runs.""" + assert True From 9b18b7f8d22325d82c0f60b5a786d3ba6b718c0d Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 14:22:27 -0400 Subject: [PATCH 07/27] Add more docstrings to search.py --- usajobsapi/client.py | 4 ++-- usajobsapi/endpoints/search.py | 22 +++++++++++++++++----- 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/usajobsapi/client.py b/usajobsapi/client.py index ce96f8c..ad2fe35 100644 --- a/usajobsapi/client.py +++ b/usajobsapi/client.py @@ -118,8 +118,8 @@ def job_search(self, **kwargs) -> SearchEndpoint.Response: """ params = SearchEndpoint.Params(**kwargs) resp = self._request( - SearchEndpoint.model_fields["method"].default, - SearchEndpoint.model_fields["path"].default, + SearchEndpoint.model_fields["METHOD"].default, + SearchEndpoint.model_fields["PATH"].default, params.to_params(), add_auth=True, ) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index fdb2f5e..37e62f7 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -18,6 +18,8 @@ class SortField(StrEnum): + """Sort the search by the specified field.""" + OPEN_DATE = "opendate" CLOSED_DATE = "closedate" ORGANIZATION_NAME = "organizationname" @@ -35,22 +37,30 @@ class SortField(StrEnum): class SortDirection(StrEnum): + """Sort the search by the SortField specified, in the direction specified.""" + ASC = "Asc" DESC = "Desc" class WhoMayApply(StrEnum): + """Filter the search by the specified candidate designation.""" + ALL = "All" PUBLIC = "Public" STATUS = "Status" class Fields(StrEnum): - MIN = "Min" + """Return the minimum or maximum number of fields for each result item.""" + + MIN = "Min" # Return only the job summary FULL = "Full" class HiringPath(StrEnum): + """Filter search results by the specified hiring path(s).""" + PUBLIC = "public" VET = "vet" N_GUARD = "nguard" @@ -74,7 +84,7 @@ class HiringPath(StrEnum): # --- class SearchEndpoint(BaseModel): """ - Declarative endpoint definition for the Job Search API. + Declarative endpoint definition for the [Job Search API](https://developer.usajobs.gov/api-reference/get-api-search). Includes the endpoint's: @@ -83,11 +93,11 @@ class SearchEndpoint(BaseModel): - Metadata """ - method: str = "GET" - path: str = "/api/search" + METHOD: str = "GET" + PATH: str = "/api/search" class Params(BaseModel): - """Declarative query-parameter model""" + """Declarative definition for the endpoint's query parameters.""" model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) @@ -243,6 +253,8 @@ def jobs(self) -> List["SearchEndpoint.JobSummary"]: return out class Response(BaseModel): + """Declarative definition for the endpoint's response object.""" + language: Optional[str] = Field(default=None, alias="LanguageCode") params: Optional["SearchEndpoint.Params"] = Field( default=None, alias="SearchParameters" From 252f2ca1d76bfe954e58bf605da0652a0fe042e7 Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 14:52:24 -0400 Subject: [PATCH 08/27] Create tests/conftest.py with SearchEndpoint fixtures --- tests/conftest.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index 56d266a..f4da9ce 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -21,6 +21,20 @@ def historicjoa_params_kwargs() -> Dict[str, str]: } +@pytest.fixture +def job_summary_payload(): + """Sample payload matching the API's job summary schema.""" + return { + "MatchedObjectId": "1", + "PositionTitle": "Engineer", + "OrganizationName": "NASA", + "PositionLocationDisplay": "Houston, TX", + "MinimumRange": 50000.0, + "MaximumRange": 100000.0, + "ApplicationCloseDate": "2024-01-01", + } + + @pytest.fixture def historicjoa_response_payload() -> Dict[str, object]: """Serialized Historic JOA response payload mimicking the USAJOBS API.""" @@ -137,3 +151,9 @@ def _historicjoa_items() -> List[Dict[str, object]]: ], }, ] + + +@pytest.fixture +def search_result_item(job_summary_payload): + """Wrap the job summary payload under the expected descriptor key.""" + return {"MatchedObjectDescriptor": job_summary_payload} From d020f83f5a1163dc11e7d87406c8e4d6f015237e Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 15:10:34 -0400 Subject: [PATCH 09/27] Import future annotation for self type hints --- usajobsapi/endpoints/search.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 37e62f7..7477e2d 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -1,5 +1,7 @@ """Wrapper for the Job Search API.""" +from __future__ import annotations + from enum import StrEnum from typing import Annotated, Any, Dict, List, Optional @@ -240,7 +242,7 @@ class SearchResult(BaseModel): alias="SearchResultItems", ) - def jobs(self) -> List["SearchEndpoint.JobSummary"]: + def jobs(self) -> List[SearchEndpoint.JobSummary]: # Convert to a normalized list versus keeping raw out: List[SearchEndpoint.JobSummary] = [] for item in self.items: @@ -256,10 +258,10 @@ class Response(BaseModel): """Declarative definition for the endpoint's response object.""" language: Optional[str] = Field(default=None, alias="LanguageCode") - params: Optional["SearchEndpoint.Params"] = Field( + params: Optional[SearchEndpoint.Params] = Field( default=None, alias="SearchParameters" ) # Results are wrapped under SearchResult - search_result: Optional["SearchEndpoint.SearchResult"] = Field( + search_result: Optional[SearchEndpoint.SearchResult] = Field( default=None, alias="SearchResult" ) From 035dc3685c2551b042678828e6ba1c320d28574d Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 15:11:28 -0400 Subject: [PATCH 10/27] Add unit tests for search.py: --- tests/unit/test_search.py | 78 +++++++++++++++++++++++++++++++++++++-- 1 file changed, 74 insertions(+), 4 deletions(-) diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index 877c54c..36eed45 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -1,6 +1,76 @@ -"""Placeholder tests for USAJobs API package.""" +import pytest +from usajobsapi.endpoints.search import HiringPath, SearchEndpoint -def test_placeholder() -> None: - """A trivial test to ensure the test suite runs.""" - assert True + +class TestSearchEndpointParams: + def test_to_params_serialization(self): + data = { + "keyword": "developer", + "location_names": ["City, ST", "Town, ST2"], + "radius": 25, + "relocation": True, + "job_category_codes": ["001", "002"], + "hiring_paths": [HiringPath.PUBLIC, HiringPath.VET], + } + params = SearchEndpoint.Params.model_validate(data) + assert params.to_params() == { + "Keyword": "developer", + "LocationName": "City, ST;Town, ST2", + "Radius": "25", + "RelocationIndicator": "True", + "JobCategoryCode": "001;002", + "HiringPath": "public;vet", + } + + def test_radius_requires_location(self): + with pytest.raises(ValueError): + SearchEndpoint.Params.model_validate({"radius": 10}) + + def test_remuneration_max_less_than_min(self): + with pytest.raises(ValueError): + SearchEndpoint.Params.model_validate( + {"remuneration_min": 100, "remuneration_max": 50} + ) + + +class TestSearchEndpointResponses: + def test_search_result_jobs_parsing(self, search_result_item): + items = [ + search_result_item, + {"MatchedObjectId": "2", "PositionTitle": "Analyst"}, + {"MatchedObjectDescriptor": {"MatchedObjectId": "3"}}, # invalid + ] + search_result = SearchEndpoint.SearchResult.model_validate( + { + "SearchResultCount": 3, + "SearchResultCountAll": 3, + "SearchResultItems": items, + } + ) + jobs = search_result.jobs() + assert len(jobs) == 2 + assert jobs[0].id == "1" + assert jobs[1].id == "2" + + def test_response_model_parsing(self, search_result_item): + response = SearchEndpoint.Response.model_validate( + { + "LanguageCode": "EN", + "SearchParameters": { + "keyword": "python", + "location_names": ["Anywhere"], + }, + "SearchResult": { + "SearchResultCount": 1, + "SearchResultCountAll": 1, + "SearchResultItems": [search_result_item], + }, + } + ) + assert response.language == "EN" + assert response.params is not None + assert response.params.keyword == "python" + assert response.search_result is not None + jobs = response.search_result.jobs() + assert jobs[0].id == "1" From f98f725662be64907e690dc3ca61527395d958e8 Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 15:50:14 -0400 Subject: [PATCH 11/27] Change client 'job_search' to 'search_jobs' --- usajobsapi/client.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/usajobsapi/client.py b/usajobsapi/client.py index ad2fe35..91df9d4 100644 --- a/usajobsapi/client.py +++ b/usajobsapi/client.py @@ -110,10 +110,10 @@ def announcement_text(self, **kwargs) -> AnnouncementTextEndpoint.Response: ) return AnnouncementTextEndpoint.Response.model_validate(resp.json()) - def job_search(self, **kwargs) -> SearchEndpoint.Response: - """Query the Job Search API. + def search_jobs(self, **kwargs) -> SearchEndpoint.Response: + """Query the [Job Search API](https://developer.usajobs.gov/api-reference/get-api-search). - :return: _description_ + :return: Active job listings matching the search criteria. :rtype: SearchEndpoint.Response """ params = SearchEndpoint.Params(**kwargs) From d4522c68a9d17f7638d7372c6e3e37bd431de43e Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 15:51:18 -0400 Subject: [PATCH 12/27] Rename USAJobsApiClient to USAJobsClient --- usajobsapi/__init__.py | 4 ++-- usajobsapi/client.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/usajobsapi/__init__.py b/usajobsapi/__init__.py index 1d36cba..cfa267c 100644 --- a/usajobsapi/__init__.py +++ b/usajobsapi/__init__.py @@ -1,6 +1,6 @@ """Top-level package for the USAJOBS REST API wrapper.""" from usajobsapi._version import __license__, __title__ -from usajobsapi.client import USAJobsApiClient +from usajobsapi.client import USAJobsClient -__all__: list[str] = ["__license__", "__title__", "USAJobsApiClient"] +__all__: list[str] = ["__license__", "__title__", "USAJobsClient"] diff --git a/usajobsapi/client.py b/usajobsapi/client.py index 91df9d4..b03ef76 100644 --- a/usajobsapi/client.py +++ b/usajobsapi/client.py @@ -13,7 +13,7 @@ ) -class USAJobsApiClient: +class USAJobsClient: """Represents a client connection to the USAJOBS REST API.""" def __init__( From c9718737ed81107069ee1bf4f301c534c5ab7489 Mon Sep 17 00:00:00 2001 From: Patrick Young Date: Thu, 11 Sep 2025 16:05:37 -0400 Subject: [PATCH 13/27] Update --- README.md | 34 ++++++++++++++++++++++++---------- 1 file changed, 24 insertions(+), 10 deletions(-) diff --git a/README.md b/README.md index 5afa553..b75edfb 100644 --- a/README.md +++ b/README.md @@ -1,12 +1,20 @@ # python-usajobsapi -A Python wrapper for the [USAJOBS REST API](https://developer.usajobs.gov/). The library aims to provide a simple interface for discovering and querying job postings from USAJOBS using Python. +A Python wrapper for the [USAJOBS REST API](https://developer.usajobs.gov/). The package aims to provide a simple SDK interface for discovering and querying job postings from USAJOBS using Python. ## Features -- Lightweight client for the USAJOBS REST endpoints -- Easily search for job postings with familiar Python types -- No external dependencies required +- Lightweight client for the USAJOBS REST API endpoints +- Leverage type hinting and validation for endpoint parameters +- Map endpoint results to Python objects + +### Supported Endpoints + +This package primarily aims to support searching and retrieval of active and past job listings. However, updates are planned to add support for all other documented endpoints. + +Currently, the following endpoints are supported: + +- [Job Search API](https://developer.usajobs.gov/api-reference/get-api-search) (`/api/Search`) ## Installation @@ -16,6 +24,12 @@ A Python wrapper for the [USAJOBS REST API](https://developer.usajobs.gov/). The pip install python-usajobsapi ``` +or, with [astral-uv](https://docs.astral.sh/uv/): + +```bash +uv add python-usajobsapi +``` + ### From source ```bash @@ -29,12 +43,12 @@ pip install . Register for a USAJOBS API key and set a valid User-Agent before making requests. ```python -from usajobsapi import USAJobs +from usajobsapi import USAJobsClient -client = USAJobs(user_agent="name@example.com", api_key="YOUR_API_KEY") -results = client.search_jobs(keyword="data scientist", location="Remote") +client = USAJobsClient(auth_user="name@example.com", auth_key="YOUR_API_KEY") +results = client.search_jobs(keyword="data scientist", location_names=["Atlanta", "Georgia"]).search_result.jobs() for job in results: - print(job["Title"]) + print(job.position_title) ``` ## Contributing @@ -50,11 +64,11 @@ Please open an issue first for major changes to discuss your proposal. ## License -Distributed under the GNU General Public License v3.0. See [LICENSE](LICENSE) for details. +Distributed under the [GNU General Public License v3.0](https://www.gnu.org/licenses/gpl-3.0.en.html). See [LICENSE](LICENSE) for details. ## Project Status -This project is under active development and the API may change. Feedback and ideas are appreciated. +This project is under active development and its API may change. Changes to the [USAJOBS REST API documentation](https://developer.usajobs.gov/) shall be monitored and incorporated into this project in a reasonable amount of time. Feedback and ideas are appreciated. ## Contact From 7f8cd16813173d35b191877bb462b14a6f8bda52 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Sun, 21 Sep 2025 01:43:36 -0400 Subject: [PATCH 14/27] Docstring all SearchEndpoint classes and funcs --- usajobsapi/endpoints/search.py | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 7477e2d..a56bffe 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -86,20 +86,14 @@ class HiringPath(StrEnum): # --- class SearchEndpoint(BaseModel): """ - Declarative endpoint definition for the [Job Search API](https://developer.usajobs.gov/api-reference/get-api-search). - - Includes the endpoint's: - - - Parameters - - Response shapes - - Metadata + Declarative wrapper around the [Job Search API](https://developer.usajobs.gov/api-reference/get-api-search). """ METHOD: str = "GET" PATH: str = "/api/search" class Params(BaseModel): - """Declarative definition for the endpoint's query parameters.""" + """Declarative definition of the endpoint's query parameters.""" model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) @@ -195,6 +189,7 @@ class Params(BaseModel): @model_validator(mode="after") def _radius_requires_location(self) -> "SearchEndpoint.Params": + """Only use radius filters when a locaiton is provided.""" if self.radius is not None and not self.location_names: raise ValueError("Radius requires at least one LocationName.") return self @@ -202,6 +197,7 @@ def _radius_requires_location(self) -> "SearchEndpoint.Params": @field_validator("remuneration_max") @classmethod def _check_min_le_max(cls, v, info): + """Validate that renumeration max is >= renumeration min.""" mn = info.data.get("remuneration_min") if v is not None and mn is not None and v < mn: raise ValueError( @@ -210,12 +206,15 @@ def _check_min_le_max(cls, v, info): return v def to_params(self) -> Dict[str, str]: + """Return the serialized query-parameter dictionary.""" return _dump_by_alias(self) # Response shapes # --- class JobSummary(BaseModel): + """Normalized representation of a search result item.""" + id: str = Field(alias="MatchedObjectId") position_title: str = Field(alias="PositionTitle") organization_name: Optional[str] = Field(default=None, alias="OrganizationName") @@ -229,6 +228,8 @@ class JobSummary(BaseModel): ) class SearchResult(BaseModel): + """Model of paginated search results.""" + result_count: Optional[int] = Field( default=None, alias="SearchResultCount", @@ -243,7 +244,7 @@ class SearchResult(BaseModel): ) def jobs(self) -> List[SearchEndpoint.JobSummary]: - # Convert to a normalized list versus keeping raw + """Normalize the list of search results, skiping malformed payloads.""" out: List[SearchEndpoint.JobSummary] = [] for item in self.items: # Some responses nest the item under 'MatchedObjectDescriptor' @@ -255,7 +256,7 @@ def jobs(self) -> List[SearchEndpoint.JobSummary]: return out class Response(BaseModel): - """Declarative definition for the endpoint's response object.""" + """Declarative definition of the endpoint's response object.""" language: Optional[str] = Field(default=None, alias="LanguageCode") params: Optional[SearchEndpoint.Params] = Field( From 922083fed6ea99bd6e120aebec48163a4e90ab64 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Sun, 21 Sep 2025 01:49:52 -0400 Subject: [PATCH 15/27] Add ValidInfo ValidErr type hints to search.py --- usajobsapi/endpoints/search.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index a56bffe..d7c30b9 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -9,6 +9,8 @@ BaseModel, ConfigDict, Field, + ValidationError, + ValidationInfo, field_validator, model_validator, ) @@ -196,7 +198,9 @@ def _radius_requires_location(self) -> "SearchEndpoint.Params": @field_validator("remuneration_max") @classmethod - def _check_min_le_max(cls, v, info): + def _check_min_le_max( + cls, v: Optional[int], info: ValidationInfo + ) -> Optional[int]: """Validate that renumeration max is >= renumeration min.""" mn = info.data.get("remuneration_min") if v is not None and mn is not None and v < mn: @@ -251,7 +255,7 @@ def jobs(self) -> List[SearchEndpoint.JobSummary]: descriptor = item.get("MatchedObjectDescriptor") or item try: out.append(SearchEndpoint.JobSummary.model_validate(descriptor)) - except Exception: + except ValidationError: continue return out From e77a19362a8746b5a32223bf87f499668a50849d Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Sun, 21 Sep 2025 01:58:53 -0400 Subject: [PATCH 16/27] Create job search result helper func --- tests/unit/test_search.py | 17 +++++++++++++++++ usajobsapi/endpoints/search.py | 6 ++++++ 2 files changed, 23 insertions(+) diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index 36eed45..8fbac67 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -74,3 +74,20 @@ def test_response_model_parsing(self, search_result_item): assert response.search_result is not None jobs = response.search_result.jobs() assert jobs[0].id == "1" + + def test_response_jobs_helper(self, search_result_item): + response = SearchEndpoint.Response.model_validate( + { + "SearchResult": { + "SearchResultCount": 1, + "SearchResultCountAll": 1, + "SearchResultItems": [search_result_item], + } + } + ) + jobs = response.jobs() + assert len(jobs) == 1 + assert jobs[0].id == "1" + + empty_response = SearchEndpoint.Response.model_validate({}) + assert empty_response.jobs() == [] diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index d7c30b9..7238688 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -270,3 +270,9 @@ class Response(BaseModel): search_result: Optional[SearchEndpoint.SearchResult] = Field( default=None, alias="SearchResult" ) + + def jobs(self) -> List[SearchEndpoint.JobSummary]: + """Helper method to directly expose parsed jobs in the response object.""" + if not self.search_result: + return [] + return self.search_result.jobs() From bb4a119a367cd92c6e3c21a3ce93b294dc744bf3 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Sun, 21 Sep 2025 02:14:04 -0400 Subject: [PATCH 17/27] Add remaining SearchEndpoint response shapes --- tests/conftest.py | 37 ++++++++ tests/unit/test_search.py | 32 +++++++ usajobsapi/endpoints/search.py | 168 +++++++++++++++++++++++++++++++++ 3 files changed, 237 insertions(+) diff --git a/tests/conftest.py b/tests/conftest.py index f4da9ce..6b34216 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,12 +26,49 @@ def job_summary_payload(): """Sample payload matching the API's job summary schema.""" return { "MatchedObjectId": "1", + "PositionID": "24-123456", "PositionTitle": "Engineer", "OrganizationName": "NASA", + "DepartmentName": "National Aeronautics and Space Administration", + "PositionURI": "https://example.com/job/1", + "ApplyURI": ["https://example.com/apply/1"], "PositionLocationDisplay": "Houston, TX", + "PositionLocation": [ + { + "LocationName": "Houston, Texas", + "LocationCode": "TX1234", + "CountryCode": "US", + "CountrySubDivisionCode": "TX", + "CityName": "Houston", + "Latitude": "29.7604", + "Longitude": "-95.3698", + } + ], + "JobCategory": [{"Code": "0801", "Name": "General Engineering"}], + "JobGrade": [{"Code": "GS", "CurrentGrade": "12"}], + "PositionSchedule": [{"Code": "1", "Name": "Full-time"}], + "PositionOfferingType": [{"Code": "15317", "Name": "Permanent"}], "MinimumRange": 50000.0, "MaximumRange": 100000.0, + "PositionRemuneration": [ + { + "MinimumRange": "50000", + "MaximumRange": "100000", + "RateIntervalCode": "PA", + "Description": "Per Year", + } + ], "ApplicationCloseDate": "2024-01-01", + "UserArea": { + "Details": { + "JobSummary": "Design and build spacecraft components.", + "HiringPath": "public;vet", + "WhoMayApply": { + "Name": "Open to the public", + "Code": "public", + }, + } + }, } diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index 8fbac67..80f42bf 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -35,6 +35,38 @@ def test_remuneration_max_less_than_min(self): class TestSearchEndpointResponses: + def test_job_summary_parses_nested_fields(self, job_summary_payload): + summary = SearchEndpoint.JobSummary.model_validate(job_summary_payload) + + assert summary.position_id == "24-123456" + assert summary.position_uri == "https://example.com/job/1" + assert summary.apply_uri == ["https://example.com/apply/1"] + assert ( + summary.department_name == "National Aeronautics and Space Administration" + ) + assert summary.locations_display == "Houston, TX" + + assert len(summary.locations) == 1 + location = summary.locations[0] + assert location.city_name == "Houston" + assert location.state_code == "TX" + assert location.latitude == pytest.approx(29.7604) + assert location.longitude == pytest.approx(-95.3698) + + assert summary.job_categories[0].code == "0801" + assert summary.job_grades[0].current_grade == "12" + assert summary.position_schedules[0].name == "Full-time" + assert summary.position_offerings[0].code == "15317" + + assert summary.salary_range() == (50000.0, 100000.0) + assert summary.hiring_paths() == ["public", "vet"] + assert summary.summary() == "Design and build spacecraft components." + + assert summary.user_area + assert summary.user_area.details + assert summary.user_area.details.who_may_apply + assert summary.user_area.details.who_may_apply.name == "Open to the public" + def test_search_result_jobs_parsing(self, search_result_item): items = [ search_result_item, diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 7238688..dd853a9 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -216,21 +216,189 @@ def to_params(self) -> Dict[str, str]: # Response shapes # --- + class JobCategory(BaseModel): + """Represents a job series classification associated with a posting.""" + + code: Optional[str] = Field(default=None, alias="Code") + name: Optional[str] = Field(default=None, alias="Name") + + class JobGrade(BaseModel): + """Represents the job grade (e.g. GS) tied to the posting.""" + + code: Optional[str] = Field(default=None, alias="Code") + current_grade: Optional[str] = Field(default=None, alias="CurrentGrade") + + class PositionSchedule(BaseModel): + """Represents the work schedule for the position.""" + + code: Optional[str] = Field(default=None, alias="Code") + name: Optional[str] = Field(default=None, alias="Name") + + class PositionOfferingType(BaseModel): + """Represents the appointment type (e.g. permanent, term).""" + + code: Optional[str] = Field(default=None, alias="Code") + name: Optional[str] = Field(default=None, alias="Name") + + class PositionRemuneration(BaseModel): + """Represents a salary range entry for the position.""" + + minimum: Optional[float] = Field(default=None, alias="MinimumRange") + maximum: Optional[float] = Field(default=None, alias="MaximumRange") + rate_interval_code: Optional[str] = Field( + default=None, alias="RateIntervalCode" + ) + description: Optional[str] = Field(default=None, alias="Description") + + @field_validator("minimum", "maximum", mode="before") + @classmethod + def _normalize_amount(cls, value: Any) -> Optional[float]: + """Normalize remuneration amounts to floats.""" + + if value in (None, ""): + return None + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, str): + cleaned = value.replace(",", "").replace("$", "").strip() + if not cleaned: + return None + try: + return float(cleaned) + except ValueError: + return None + return None + + class JobLocation(BaseModel): + """Represents a structured job location entry.""" + + name: Optional[str] = Field(default=None, alias="LocationName") + code: Optional[str] = Field(default=None, alias="LocationCode") + country_code: Optional[str] = Field(default=None, alias="CountryCode") + state_code: Optional[str] = Field(default=None, alias="CountrySubDivisionCode") + city_name: Optional[str] = Field(default=None, alias="CityName") + latitude: Optional[float] = Field(default=None, alias="Latitude") + longitude: Optional[float] = Field(default=None, alias="Longitude") + + @field_validator("latitude", "longitude", mode="before") + @classmethod + def _normalize_coordinate(cls, value: Any) -> Optional[float]: + """Normalize coordinate values to floats.""" + + if value in (None, ""): + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + class WhoMayApplyInfo(BaseModel): + """Represents the structured WhoMayApply block.""" + + name: Optional[str] = Field(default=None, alias="Name") + code: Optional[str] = Field(default=None, alias="Code") + + class UserAreaDetails(BaseModel): + """Represents metadata stored under the UserArea.Details field.""" + + job_summary: Optional[str] = Field(default=None, alias="JobSummary") + hiring_path: Optional[str] = Field(default=None, alias="HiringPath") + who_may_apply: Optional[SearchEndpoint.WhoMayApplyInfo] = Field( + default=None, alias="WhoMayApply" + ) + + class UserArea(BaseModel): + """Wrapper for additional USAJOBS metadata.""" + + details: Optional[SearchEndpoint.UserAreaDetails] = Field( + default=None, alias="Details" + ) + class JobSummary(BaseModel): """Normalized representation of a search result item.""" id: str = Field(alias="MatchedObjectId") + position_id: Optional[str] = Field(default=None, alias="PositionID") position_title: str = Field(alias="PositionTitle") + position_uri: Optional[str] = Field(default=None, alias="PositionURI") + # URI to apply for the job offering + apply_uri: List[str] = Field(default_factory=list, alias="ApplyURI") organization_name: Optional[str] = Field(default=None, alias="OrganizationName") + department_name: Optional[str] = Field(default=None, alias="DepartmentName") locations_display: Optional[str] = Field( default=None, alias="PositionLocationDisplay" ) + locations: List[SearchEndpoint.JobLocation] = Field( + default_factory=list, alias="PositionLocation" + ) + job_categories: List[SearchEndpoint.JobCategory] = Field( + default_factory=list, alias="JobCategory" + ) + job_grades: List[SearchEndpoint.JobGrade] = Field( + default_factory=list, alias="JobGrade" + ) + position_schedules: List[SearchEndpoint.PositionSchedule] = Field( + default_factory=list, alias="PositionSchedule" + ) + position_offerings: List[SearchEndpoint.PositionOfferingType] = Field( + default_factory=list, alias="PositionOfferingType" + ) + user_area: Optional[SearchEndpoint.UserArea] = Field( + default=None, alias="UserArea" + ) + qualification_summary: Optional[str] = Field( + default=None, alias="QualificationSummary" + ) min_salary: Optional[float] = Field(default=None, alias="MinimumRange") max_salary: Optional[float] = Field(default=None, alias="MaximumRange") + position_remuneration: List[SearchEndpoint.PositionRemuneration] = Field( + default_factory=list, alias="PositionRemuneration" + ) + publication_start_date: Optional[str] = Field( + default=None, alias="PublicationStartDate" + ) application_close_date: Optional[str] = Field( default=None, alias="ApplicationCloseDate" ) + def summary(self) -> Optional[str]: + """Helper method returning the most descriptive summary for the job.""" + + if self.user_area and self.user_area.details: + details = self.user_area.details + if details and details.job_summary: + return details.job_summary + return self.qualification_summary + + def salary_range(self) -> tuple[Optional[float], Optional[float]]: + """Helper method returning the salary range inferred from remuneration metadata.""" + + if self.position_remuneration: + minima = [ + r.minimum + for r in self.position_remuneration + if r.minimum is not None + ] + maxima = [ + r.maximum + for r in self.position_remuneration + if r.maximum is not None + ] + min_val = min(minima) if minima else None + max_val = max(maxima) if maxima else None + if min_val is not None or max_val is not None: + return min_val, max_val + return self.min_salary, self.max_salary + + def hiring_paths(self) -> List[str]: + """Helper method returning normalized hiring path codes from the user area.""" + + if self.user_area and self.user_area.details: + raw = self.user_area.details.hiring_path + if raw: + return [path.strip() for path in raw.split(";") if path.strip()] + return [] + class SearchResult(BaseModel): """Model of paginated search results.""" From fc3c6521d547722ed826b495d79c2827cc9957d2 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Fri, 26 Sep 2025 12:53:20 -0400 Subject: [PATCH 18/27] Fix rebase conflicts --- tests/conftest.py | 48 +++++++++++++++++++-------------------- tests/unit/test_client.py | 16 ++++++------- 2 files changed, 32 insertions(+), 32 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 6b34216..42df889 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,24 +3,6 @@ import pytest -@pytest.fixture -def historicjoa_params_kwargs() -> Dict[str, str]: - """Field-value mapping used to build HistoricJoaEndpoint params models.""" - - return { - "hiring_agency_codes": "AGENCY1", - "hiring_department_codes": "DEPT1", - "position_series": "2210", - "announcement_numbers": "23-ABC", - "usajobs_control_numbers": "1234567", - "start_position_open_date": "2020-01-01", - "end_position_open_date": "2020-12-31", - "start_position_close_date": "2021-01-01", - "end_position_close_date": "2021-12-31", - "continuation_token": "token123", - } - - @pytest.fixture def job_summary_payload(): """Sample payload matching the API's job summary schema.""" @@ -72,6 +54,30 @@ def job_summary_payload(): } +@pytest.fixture +def search_result_item(job_summary_payload): + """Wrap the job summary payload under the expected descriptor key.""" + return {"MatchedObjectDescriptor": job_summary_payload} + + +@pytest.fixture +def historicjoa_params_kwargs() -> Dict[str, str]: + """Field-value mapping used to build HistoricJoaEndpoint params models.""" + + return { + "hiring_agency_codes": "AGENCY1", + "hiring_department_codes": "DEPT1", + "position_series": "2210", + "announcement_numbers": "23-ABC", + "usajobs_control_numbers": "1234567", + "start_position_open_date": "2020-01-01", + "end_position_open_date": "2020-12-31", + "start_position_close_date": "2021-01-01", + "end_position_close_date": "2021-12-31", + "continuation_token": "token123", + } + + @pytest.fixture def historicjoa_response_payload() -> Dict[str, object]: """Serialized Historic JOA response payload mimicking the USAJOBS API.""" @@ -188,9 +194,3 @@ def _historicjoa_items() -> List[Dict[str, object]]: ], }, ] - - -@pytest.fixture -def search_result_item(job_summary_payload): - """Wrap the job summary payload under the expected descriptor key.""" - return {"MatchedObjectDescriptor": job_summary_payload} diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 88565e7..a9d2625 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,10 +1,10 @@ -"""Unit tests for USAJobsApiClient.""" +"""Unit tests for USAJobsClient.""" from copy import deepcopy import pytest -from usajobsapi.client import USAJobsApiClient +from usajobsapi.client import USAJobsClient from usajobsapi.endpoints.historicjoa import HistoricJoaEndpoint # test historic_joa_pages @@ -31,9 +31,9 @@ def fake_historic_joa(self, **call_kwargs): captured_kwargs.append(call_kwargs) return responses.pop(0) - monkeypatch.setattr(USAJobsApiClient, "historic_joa", fake_historic_joa) + monkeypatch.setattr(USAJobsClient, "historic_joa", fake_historic_joa) - client = USAJobsApiClient() + client = USAJobsClient() pages = list( client.historic_joa_pages( @@ -70,9 +70,9 @@ def test_historic_joa_pages_duplicate_token( def fake_historic_joa(self, **_): return responses.pop(0) - monkeypatch.setattr(USAJobsApiClient, "historic_joa", fake_historic_joa) + monkeypatch.setattr(USAJobsClient, "historic_joa", fake_historic_joa) - client = USAJobsApiClient() + client = USAJobsClient() iterator = client.historic_joa_pages() assert next(iterator) @@ -89,7 +89,7 @@ def test_historic_joa_items_yields_items_across_pages( ) -> None: """Ensure historic_joa_items yields items and follows continuation tokens.""" - client = USAJobsApiClient() + client = USAJobsClient() first_page = deepcopy(historicjoa_response_payload) first_page["paging"]["metadata"]["continuationToken"] = "TOKEN2" @@ -178,7 +178,7 @@ def test_historic_joa_items_respects_initial_token( ) -> None: """Ensure historic_joa_items uses the supplied initial continuation token.""" - client = USAJobsApiClient() + client = USAJobsClient() payload = deepcopy(historicjoa_response_payload) payload["paging"]["metadata"]["continuationToken"] = None From b4a5afa0f970c7b2c30fdc9e05115697cb8d2503 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Fri, 26 Sep 2025 15:53:51 -0400 Subject: [PATCH 19/27] Add generator for job search pages --- tests/unit/test_client.py | 114 ++++++++++++++++++++++++++++++++++++++ usajobsapi/client.py | 59 ++++++++++++++++++++ 2 files changed, 173 insertions(+) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index a9d2625..ba331bc 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -6,6 +6,120 @@ from usajobsapi.client import USAJobsClient from usajobsapi.endpoints.historicjoa import HistoricJoaEndpoint +from usajobsapi.endpoints.search import SearchEndpoint + +# test search_jobs_pages +# --- + + +def _build_search_payload(items, count, total=None): + """Build a serialized SearchResult payload for mocked responses.""" + + payload = { + "SearchResult": { + "SearchResultCount": count, + "SearchResultItems": items, + } + } + if total is not None: + payload["SearchResult"]["SearchResultCountAll"] = total + return payload + + +def test_search_jobs_pages_yields_pages(monkeypatch, search_result_item) -> None: + """Ensure search_jobs_pages iterates pages based on total counts.""" + + first_item = deepcopy(search_result_item) + first_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "1" + second_item = deepcopy(search_result_item) + second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "2" + third_item = deepcopy(search_result_item) + third_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "3" + fourth_item = deepcopy(search_result_item) + fourth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "4" + fifth_item = deepcopy(search_result_item) + fifth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "5" + + responses = [ + SearchEndpoint.Response.model_validate( + _build_search_payload([first_item, second_item], 2, total=5) + ), + SearchEndpoint.Response.model_validate( + _build_search_payload([third_item, fourth_item], 2, total=5) + ), + SearchEndpoint.Response.model_validate( + _build_search_payload([fifth_item], 1, total=5) + ), + ] + + captured_kwargs = [] + + def fake_search(self, **call_kwargs): + captured_kwargs.append(call_kwargs) + return responses.pop(0) + + monkeypatch.setattr(USAJobsClient, "search_jobs", fake_search) + + client = USAJobsClient() + pages = list(client.search_jobs_pages(keyword="engineer", results_per_page=2)) + + assert len(pages) == 3 + assert [call["page"] for call in captured_kwargs] == [1, 2, 3] + assert all(call["results_per_page"] == 2 for call in captured_kwargs) + + +def test_search_jobs_pages_handles_missing_total( + monkeypatch, search_result_item +) -> None: + """Continue until a short page is returned when total counts are absent.""" + + first_page = SearchEndpoint.Response.model_validate( + _build_search_payload( + [deepcopy(search_result_item), deepcopy(search_result_item)], + 2, + ) + ) + second_item = deepcopy(search_result_item) + second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "3" + second_page = SearchEndpoint.Response.model_validate( + _build_search_payload([second_item], 1) + ) + + responses = [first_page, second_page] + captured_kwargs = [] + + def fake_search(self, **call_kwargs): + captured_kwargs.append(call_kwargs) + return responses.pop(0) + + monkeypatch.setattr(USAJobsClient, "search_jobs", fake_search) + + client = USAJobsClient() + pages = list(client.search_jobs_pages(keyword="space")) + + assert len(pages) == 2 + assert [call["page"] for call in captured_kwargs] == [1, 2] + assert "results_per_page" not in captured_kwargs[0] + assert captured_kwargs[1]["results_per_page"] == 2 + + +def test_search_jobs_pages_breaks_on_empty_results(monkeypatch) -> None: + """Stop pagination when a page returns no results.""" + + empty_page = SearchEndpoint.Response.model_validate( + {"SearchResult": {"SearchResultCount": 0, "SearchResultItems": []}} + ) + + def fake_search(self, **_): + return empty_page + + monkeypatch.setattr(USAJobsClient, "search_jobs", fake_search) + + client = USAJobsClient() + pages = list(client.search_jobs_pages(keyword="empty")) + + assert len(pages) == 1 + # test historic_joa_pages # --- diff --git a/usajobsapi/client.py b/usajobsapi/client.py index b03ef76..defd21f 100644 --- a/usajobsapi/client.py +++ b/usajobsapi/client.py @@ -125,6 +125,65 @@ def search_jobs(self, **kwargs) -> SearchEndpoint.Response: ) return SearchEndpoint.Response.model_validate(resp.json()) + def search_jobs_pages(self, **kwargs) -> Iterator[SearchEndpoint.Response]: + """Yield job search result pages, paginating to the next page. + + This can handle fresh requests or continue requests from a given page number. + + :yield: The response object for the given page number. + :rtype: Iterator[SearchEndpoint.Response] + """ + + # Get page parameters by object name or alias + page_number: Optional[int] = kwargs.pop("page", kwargs.pop("Page", None)) + results_per_page = kwargs.pop( + "results_per_page", kwargs.pop("ResultsPerPage", None) + ) + + # If not provided, then start at the first page + current_page: int = page_number or 1 + + while True: + call_kwargs = kwargs + call_kwargs["page"] = current_page + + # results_per_page may not exist for the first loop iteration + if results_per_page: + call_kwargs["results_per_page"] = results_per_page + + # Query for the response object + resp = self.search_jobs(**call_kwargs) + yield resp + + # Break if no search_result object exists + search_result = resp.search_result + if not search_result: + break + + # Break if there are no search_result.items + page_results_count = search_result.result_count or len(search_result.items) + if page_results_count <= 0: + break + + # results_per_page may not exist for the first loop iteration + # so set it to the length of the returned search_result.items + if results_per_page is None: + results_per_page = page_results_count + + # Break if there are no more pages + total_result_count = search_result.result_total + if ( + total_result_count + and current_page * results_per_page >= total_result_count + ): + break + + # Break if the page is shorter than results_per_page + if page_results_count < results_per_page: + break + + current_page += 1 + def historic_joa(self, **kwargs) -> HistoricJoaEndpoint.Response: """Query the Historic JOAs API. From 05db89a55a0451ea93c9488ef5528bfb092eb81f Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Fri, 26 Sep 2025 16:07:37 -0400 Subject: [PATCH 20/27] Add generator for job search items --- tests/unit/test_client.py | 67 +++++++++++++++++++++++++++++++++++++++ usajobsapi/client.py | 12 ++++++- 2 files changed, 78 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index ba331bc..db2e082 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -121,6 +121,73 @@ def fake_search(self, **_): assert len(pages) == 1 +# test search_jobs_items +# --- + + +def _search_response_payload( + items: list[dict], + count: int, + total: int, + page: int, + results_per_page: int, +) -> dict: + return { + "LanguageCode": "EN", + "SearchParameters": { + "page": page, + "results_per_page": results_per_page, + }, + "SearchResult": { + "SearchResultCount": count, + "SearchResultCountAll": total, + "SearchResultItems": items, + }, + } + + +def test_search_jobs_items_yields_jobs(monkeypatch, job_summary_payload) -> None: + """Ensure search_jobs_items yields summaries across pages.""" + + client = USAJobsClient() + + first_payload = deepcopy(job_summary_payload) + first_payload["MatchedObjectId"] = "1" + second_payload = deepcopy(job_summary_payload) + second_payload["MatchedObjectId"] = "2" + third_payload = deepcopy(job_summary_payload) + third_payload["MatchedObjectId"] = "3" + + responses = [ + _search_response_payload( + [ + {"MatchedObjectDescriptor": first_payload}, + {"MatchedObjectDescriptor": second_payload}, + ], + 2, + 3, + 1, + 2, + ), + _search_response_payload( + [{"MatchedObjectDescriptor": third_payload}], + 1, + 3, + 2, + 2, + ), + ] + + def fake_search_jobs(self, **_): + return SearchEndpoint.Response.model_validate(responses.pop(0)) + + monkeypatch.setattr(USAJobsClient, "search_jobs", fake_search_jobs) + + summaries = list(client.search_jobs_items(results_per_page=2)) + + assert [summary.id for summary in summaries] == ["1", "2", "3"] + + # test historic_joa_pages # --- diff --git a/usajobsapi/client.py b/usajobsapi/client.py index defd21f..49c2dd8 100644 --- a/usajobsapi/client.py +++ b/usajobsapi/client.py @@ -126,7 +126,7 @@ def search_jobs(self, **kwargs) -> SearchEndpoint.Response: return SearchEndpoint.Response.model_validate(resp.json()) def search_jobs_pages(self, **kwargs) -> Iterator[SearchEndpoint.Response]: - """Yield job search result pages, paginating to the next page. + """Yield Job Search pages, paginating to the next page. This can handle fresh requests or continue requests from a given page number. @@ -184,6 +184,16 @@ def search_jobs_pages(self, **kwargs) -> Iterator[SearchEndpoint.Response]: current_page += 1 + def search_jobs_items(self, **kwargs) -> Iterator[SearchEndpoint.JobSummary]: + """Yield Job Search job items, handling pagination as needed. + + :yield: The job summary item. + :rtype: Iterator[SearchEndpoint.JobSummary] + """ + for resp in self.search_jobs_pages(**kwargs): + for item in resp.jobs(): + yield item + def historic_joa(self, **kwargs) -> HistoricJoaEndpoint.Response: """Query the Historic JOAs API. From ea8415742bab77085867e74392a82a964f8bfe3d Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Mon, 29 Sep 2025 14:44:11 -0400 Subject: [PATCH 21/27] Add is_inrange util function --- tests/unit/test_utils.py | 14 ++++++++++++++ usajobsapi/utils.py | 17 +++++++++++++++++ 2 files changed, 31 insertions(+) diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index fdb924e..1f05a3e 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -7,6 +7,7 @@ from usajobsapi.utils import ( _dump_by_alias, + _is_inrange, _normalize_date, _normalize_param, _normalize_yn_bool, @@ -221,3 +222,16 @@ def test_idempotent_and_no_side_effects(): assert out1 == out2 # ensure model fields weren't mutated assert m.a_list_str == ["p", "q"] + + +# test _dump_by_alias +# --- + + +def test_is_inrange(): + # int + assert _is_inrange(1, 0, 2) + assert not _is_inrange(8, 0, 2) + # float + assert _is_inrange(1.4, 0.5, 2.3) + assert not _is_inrange(8.6, 0.5, 2.3) diff --git a/usajobsapi/utils.py b/usajobsapi/utils.py index ef63e40..908f3c8 100644 --- a/usajobsapi/utils.py +++ b/usajobsapi/utils.py @@ -95,3 +95,20 @@ def _dump_by_alias(model: BaseModel) -> Dict[str, str]: if norm_val: out[k] = norm_val return out + + +def _is_inrange(n: int | float, lower: int | float, upper: int | float): + """A simple utility function that checks a given value is within the given closed interval. + + A closed interval [a, b] represents the set of all real numbers greater or equal to a and less or equal to b. + + :param n: _description_ + :type n: int + :param lower: _description_ + :type lower: int + :param upper: _description_ + :type upper: int + :return: _description_ + :rtype: _type_ + """ + return n >= lower and n <= upper From af6bf0f44be571acec23a7faabc8ba5c2042c35a Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Mon, 29 Sep 2025 15:21:29 -0400 Subject: [PATCH 22/27] Add desc and validators to SearchEndpoint.Params --- usajobsapi/endpoints/_validators.py | 19 ++ usajobsapi/endpoints/search.py | 292 +++++++++++++++++++++++----- 2 files changed, 257 insertions(+), 54 deletions(-) create mode 100644 usajobsapi/endpoints/_validators.py diff --git a/usajobsapi/endpoints/_validators.py b/usajobsapi/endpoints/_validators.py new file mode 100644 index 0000000..7465bfa --- /dev/null +++ b/usajobsapi/endpoints/_validators.py @@ -0,0 +1,19 @@ +from typing import List + +from usajobsapi.utils import _is_inrange + + +def isvalid_pay_grade(value: str): + if value in ("01", "02", "03", "04", "05", "06", "07", "08", "09", "10"): + return value + if value in ("1", "2", "3", "4", "5", "6", "7", "8", "9"): + return f"0{value}" + raise ValueError(f"{value} must be a GS pay grade (01-15)") + + +def isvalid_pos_sensitivity(value: List[int]): + if all(_is_inrange(x, 1, 7) for x in value): + return value + raise ValueError( + "Acceptable values for Position Sensitivity and Risk parameter are 1 through 7." + ) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index dd853a9..bd44566 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -6,6 +6,7 @@ from typing import Annotated, Any, Dict, List, Optional from pydantic import ( + AfterValidator, BaseModel, ConfigDict, Field, @@ -15,6 +16,7 @@ model_validator, ) +from usajobsapi.endpoints._validators import isvalid_pay_grade from usajobsapi.utils import _dump_by_alias # Enums for query-params @@ -55,7 +57,7 @@ class WhoMayApply(StrEnum): STATUS = "Status" -class Fields(StrEnum): +class FieldsMinMax(StrEnum): """Return the minimum or maximum number of fields for each result item.""" MIN = "Min" # Return only the job summary @@ -99,94 +101,268 @@ class Params(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) - keyword: Optional[str] = Field(None, serialization_alias="Keyword") - position_title: Optional[str] = Field(None, serialization_alias="PositionTitle") + keyword: Optional[str] = Field( + None, + serialization_alias="Keyword", + description="Issues search to find hits based on a keyword. Optional. Keyword will search for all of the words specified (or synonyms of the word) throughout the job announcement.", + examples=["https://data.usajobs.gov/api/search?Keyword=Software"], + ) + position_title: Optional[str] = Field( + None, + serialization_alias="PositionTitle", + description=""" + Issues search to find hits in the title of the job. + + This is the job title - e.g. IT Specialist, Psychologist, etc. The title search will be treated as 'contains' and will select all job announcements where the job title contains the value provided.""", + examples=[ + "The following query will return all job announcements with 'psychologist' or a synonym of psychologist in the title: 'https://data.usajobs.gov/api/Search?PositionTitle=Psychologist'", + "The following query will return all job announcements with 'Electrical Engineer'' in the title: 'https://data.usajobs.gov/api/Search?PositionTitle=Electrical%20Engineer'", + ], + ) remuneration_min: Optional[int] = Field( - None, serialization_alias="RemunerationMinimumAmount" + None, + serialization_alias="RemunerationMinimumAmount", + description=""" + Issues search to find hits with the minimum salary specified. + + Jobs are placed in salary buckets: $0-$24,999, $25,000-$49,999, $50,000-$74,999, $75,000-$99,999, $100,000-$124,999, $125,000-$149,999, $150,000-$174,999, $175,000-$199,999 and $200,000 or greater. So a search with a minimum salary of $15,500 will return jobs with a minimum salary in the $0-$24,999 range.""", + examples=[ + "https://data.usajobs.gov/api/Search?RemunerationMinimumAmount=15000" + ], + ge=0, ) remuneration_max: Optional[int] = Field( - None, serialization_alias="RemunerationMaximumAmount" + None, + serialization_alias="RemunerationMaximumAmount", + description=""" + Issues search to find hits with the maximum salary specified. + + Jobs are placed in salary buckets: $0-$24,999, $25,000-$49,999, $50,000-$74,999, $75,000-$99,999, $100,000-$124,999, $125,000-$149,999, $150,000-$174,999, $175,000-$199,999 and $200,000 or greater. So a search with a maximum salary of $72,000 will return jobs with a maximum salary in the $50,000-$74,999 range. + """, + examples=[ + "https://data.usajobs.gov/api/Search?RemunerationMinimumAmount=26000&RemunerationMaximumAmount=85000" + ], + gt=0, + ) + + pay_grade_high: Annotated[Optional[str], AfterValidator(isvalid_pay_grade)] = ( + Field( + None, + serialization_alias="PayGradeHigh", + description=""" + Issues search to find hits with the maximum pay grade specified. Must be 01 through 15. This is the ending grade for the job. (Caution: Fed speak ahead but it cannot be helped.) The grade along with series is used by the Federal government to categorize and define jobs. + + For more information on what series and grade are, please visit: https://help.usajobs.gov/index.php/What_is_a_series_and_or_grade%3F. + + However, grade is also used to define salary. USAJOBS search uses grades for the General Schedule (GS) pay plan ( http://www.opm.gov/policy-data-oversight/pay-leave/salaries-wages). + + For jobs that use a different pay plan than the GS schedule, USAJOBS will derive the corresponding grade by using the minimum and maximum salary and the wages for the GS schedule for the Rest of the United States (for 2014, see: http//www.opm.gov/policy-data-oversight/pay-leave/salaries-wages/salary-tables/14Tables/html/RUS.aspx). + + For federal employees, especially those who have a GS pay plan, searching by grade is extremely useful since they would already know which grades they are or qualify for. However, for non-GS employees, searching by salary is much simpler. + """, + examples=["https://data.usajobs.gov/api/Search?PayGradeHigh=07"], + ) + ) + pay_grade_low: Annotated[Optional[str], AfterValidator(isvalid_pay_grade)] = ( + Field( + None, + serialization_alias="PayGradeLow", + description="Issues search to find hits with the minimum pay grade specified. Must be 01 through 15. This is the beginning grade for the job. See PayGradeHigh for more information.", + examples=[ + "https://data.usajobs.gov/api/Search?PayGradeLow=04", + "https://data.usajobs.gov/api/Search?PayGradeLow=07&PayGradeHigh=09", + ], + ) ) - pay_grade_high: Optional[str] = Field(None, serialization_alias="PayGradeHigh") - pay_grade_low: Optional[str] = Field(None, serialization_alias="PayGradeLow") job_category_codes: List[str] = Field( - default_factory=list, serialization_alias="JobCategoryCode" + default_factory=list, + serialization_alias="JobCategoryCode", + description="Issues a search to find hits with the job category series specified.", + examples=["https://data.usajobs.gov/api/Search?JobCategoryCode=0830"], ) - position_schedule_type_codes: List[str] = Field( - default_factory=list, serialization_alias="PositionScheduleTypeCode" + position_schedule_type_codes: List[int] = Field( + default_factory=list, + serialization_alias="PositionScheduleTypeCode", + description="Issues a search to find hits for jobs matching the specified job schedule. This field is also known as work schedule.", + examples=["https://data.usajobs.gov/api/Search?PositionSchedule=4"], ) - position_offering_type_codes: List[str] = Field( - default_factory=list, serialization_alias="PositionOfferingTypeCode" + position_offering_type_codes: List[int] = Field( + default_factory=list, + serialization_alias="PositionOfferingTypeCode", + description="Issues a search to find jobs within the specified type. This field is also known as Work Type.", ) organization: List[str] = Field( - default_factory=list, serialization_alias="Organization" + default_factory=list, + serialization_alias="Organization", + description="Issues a search to find jobs for the specified agency using the Agency Subelement Code.", + examples=["https://data.usajobs.gov/api/Search?Organization=TR"], ) + location_names: List[str] = Field( - default_factory=list, serialization_alias="LocationName" + default_factory=list, + serialization_alias="LocationName", + description="Issues a search to find hits within the specified location. This is the city or military installation name. LocationName simplifies location based search as the user does not need to know or account for each and every Location Code. LocationName will search for all location codes and ZIP codes that have that specific description.", + examples=[ + "https://data.usajobs.gov/api/Search?LocationName=Washington%20DC,%20District%20of%20Columbia", + "https://data.usajobs.gov/api/Search?LocationName=Atlanta,%20Georgia", + ], ) - radius: Annotated[ - Optional[int], Field(serialization_alias="Radius", strict=True, gt=0) - ] = None - travel_percentage: List[str] = Field( - default_factory=list, serialization_alias="TravelPercentage" + travel_percentage: List[int] = Field( + default_factory=list, + serialization_alias="TravelPercentage", + description="Issues a search to find hits for jobs matching the specified travel level.", + examples=[ + "https://data.usajobs.gov/api/Search?TravelPercentage=0", + "https://data.usajobs.gov/api/Search?TravelPercentage=7", + ], + ge=0, + le=8, ) relocation: Optional[bool] = Field( - None, serialization_alias="RelocationIndicator" - ) - security_clearance_required: List[str] = Field( - default_factory=list, serialization_alias="SecurityClearanceRequired" - ) - position_sensitivity: List[str] = Field( - default_factory=list, serialization_alias="PositionSensitivity" - ) - - who_may_apply: Optional[WhoMayApply] = Field( - None, serialization_alias="WhoMayApply" - ) - hiring_paths: List[HiringPath] = Field( - default_factory=list, serialization_alias="HiringPath" - ) - - salary_bucket: List[str] = Field( - default_factory=list, serialization_alias="SalaryBucket" + None, + serialization_alias="RelocationIndicator", + description="Issues a search to find hits for jobs matching the relocation filter.", ) - grade_bucket: List[str] = Field( - default_factory=list, serialization_alias="GradeBucket" + security_clearance_required: List[int] = Field( + default_factory=list, + serialization_alias="SecurityClearanceRequired", + description="Issues a search to find hits for jobs matching the specified security clearance.", + examples=[ + "https://data.usajobs.gov/api/Search?SecurityClearanceRequired=1" + ], + ge=0, + le=8, ) supervisory_status: Optional[str] = Field( - None, serialization_alias="SupervisoryStatus" + None, + serialization_alias="SupervisoryStatus", + description="Issues a search to find hits for jobs matching the specified supervisory status.", ) - date_posted_days: Annotated[ + + days_since_posted: Annotated[ Optional[int], - Field(serialization_alias="DatePosted", strict=True, ge=0, le=60), + Field( + serialization_alias="DatePosted", + description="Issues a search to find hits for jobs that were posted within the number of days specified.", + strict=True, + ge=0, + le=60, + ), ] = None job_grade_codes: List[str] = Field( - default_factory=list, serialization_alias="JobGradeCode" - ) - mission_critical_tags: List[str] = Field( - default_factory=list, serialization_alias="MissionCriticalTags" + default_factory=list, + serialization_alias="JobGradeCode", + description="Issues a search to find hits for jobs matching the grade code specified. This field is also known as Pay Plan.", ) - sort_field: Optional[SortField] = Field(None, serialization_alias="SortField") + sort_field: Optional[SortField] = Field( + None, + serialization_alias="SortField", + description="Issues a search that will be sorted by the specified field.", + examples=[ + "https://data.usajobs.gov/api/Search?PositionTitle=Electrical&SortField=PositionTitle" + ], + ) sort_direction: Optional[SortDirection] = Field( - None, serialization_alias="SortDirection" + None, + serialization_alias="SortDirection", + description="Issues a search that will be sorted by the SortField specified, in the direction specified.", ) + page: Annotated[ - Optional[int], Field(serialization_alias="Page", strict=True, ge=1) + Optional[int], + Field( + serialization_alias="Page", + description="Issues a search to pull the paged results specified.", + strict=True, + ge=1, + ), ] = None results_per_page: Annotated[ Optional[int], - Field(serialization_alias="ResultsPerPage", strict=True, ge=1, le=500), + Field( + serialization_alias="ResultsPerPage", + description="Issues a search and returns the page size specified. In this example, 25 jobs will be return for the first page.", + strict=True, + ge=1, + le=500, + ), + ] = None + + who_may_apply: Optional[WhoMayApply] = Field( + None, + serialization_alias="WhoMayApply", + description="Issues a search to find hits based on the desired candidate designation. In this case, public will find jobs that U.S. citizens can apply for.", + examples=["https://data.usajobs.gov/api/Search?WhoMayApply=public"], + ) + + radius: Annotated[ + Optional[int], + Field( + serialization_alias="Radius", + description="Issues a search when used along with LocationName, will expand the locations, based on the radius specified.", + examples=[ + "https://data.usajobs.gov/api/Search?LocationName=Norfolk%20Virginia&Radius=75" + ], + strict=True, + gt=0, + ), ] = None - fields: Optional[Fields] = Field(None, serialization_alias="Fields") + fields: Optional[FieldsMinMax] = Field( + None, + serialization_alias="Fields", + description="Issues a search that will return the minimum fields or maximum number of fields in the job. Min returns only the job summary.", + examples=[ + "https://data.usajobs.gov/api/Search?TravelPercentage=7&Fields=full", + "https://data.usajobs.gov/api/Search?SecurityClearanceRequired=1&Fields=full", + ], + ) + + salary_bucket: List[int] = Field( + default_factory=list, + serialization_alias="SalaryBucket", + description="Issues a search that will find hits for salaries matching the grouping specified. Buckets are assigned based on salary ranges.", + ) + grade_bucket: List[int] = Field( + default_factory=list, + serialization_alias="GradeBucket", + description="Issues a search that will find hits for grades that match the grouping specified.", + ) + + hiring_paths: List[HiringPath] = Field( + default_factory=list, + serialization_alias="HiringPath", + description="Issues a search that will find hits for hiring paths that match the hiring paths specified.", + examples=["https://data.usajobs.gov/api/Search?HiringPath=public"], + ) + + mission_critical_tags: List[str] = Field( + default_factory=list, + serialization_alias="MissionCriticalTags", + description="Issues a search that will find hits for mission critical tags that match the grouping specified.", + examples=[ + "https://data.usajobs.gov/api/Search?MissionCriticalTags=STEM&Fields=full" + ], + ) + + position_sensitivity: List[int] = Field( + default_factory=list, + serialization_alias="PositionSensitivity", + description="Issues a search that will find hits for jobs matching the position sensitivity and risk specified.", + examples=["https://data.usajobs.gov/api/Search?PositionSensitivity=1"], + ge=1, + le=7, + ) remote_indicator: Optional[bool] = Field( - None, serialization_alias="RemoteIndicator" + None, + serialization_alias="RemoteIndicator", + description="Issues a search to find hits for jobs matching the remote filter.", ) @model_validator(mode="after") @@ -405,14 +581,17 @@ class SearchResult(BaseModel): result_count: Optional[int] = Field( default=None, alias="SearchResultCount", + description="Number of records returned in response object.", ) result_total: Optional[int] = Field( default=None, alias="SearchResultCountAll", + description="Total Number of records that matched search criteria.", ) items: List[Dict[str, Any]] = Field( default_factory=list, alias="SearchResultItems", + description="Array of job opportunity announcement objects that matched search criteria.", ) def jobs(self) -> List[SearchEndpoint.JobSummary]: @@ -430,10 +609,15 @@ def jobs(self) -> List[SearchEndpoint.JobSummary]: class Response(BaseModel): """Declarative definition of the endpoint's response object.""" - language: Optional[str] = Field(default=None, alias="LanguageCode") + language: Optional[str] = Field( + default=None, alias="LanguageCode", description="Response Langauge" + ) params: Optional[SearchEndpoint.Params] = Field( - default=None, alias="SearchParameters" + default=None, + alias="SearchParameters", + description="Query parameters used in search request.", ) + # Results are wrapped under SearchResult search_result: Optional[SearchEndpoint.SearchResult] = Field( default=None, alias="SearchResult" From e8a99a03af47b9e6f67cf2dc561504924f71c744 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Mon, 29 Sep 2025 15:32:44 -0400 Subject: [PATCH 23/27] Handle SearchEndpoint dates as dt.date types --- tests/conftest.py | 24 +++++++++++++++++++++++- tests/unit/test_search.py | 22 ++++++++++------------ usajobsapi/endpoints/search.py | 18 +++++++++++++++--- 3 files changed, 48 insertions(+), 16 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 42df889..5b334c1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,7 +1,25 @@ -from typing import Dict, List +from typing import Any, Dict, List import pytest +from usajobsapi.endpoints.search import HiringPath + +# search fixtures +# --- + + +@pytest.fixture +def search_params_kwargs() -> Dict[str, Any]: + """Field-value mapping used to build SearchEndpoint params models.""" + return { + "keyword": "developer", + "location_names": ["City, ST", "Town, ST2"], + "radius": 25, + "relocation": True, + "job_category_codes": ["001", "002"], + "hiring_paths": [HiringPath.PUBLIC, HiringPath.VET], + } + @pytest.fixture def job_summary_payload(): @@ -60,6 +78,10 @@ def search_result_item(job_summary_payload): return {"MatchedObjectDescriptor": job_summary_payload} +# historicjoa fixtures +# --- + + @pytest.fixture def historicjoa_params_kwargs() -> Dict[str, str]: """Field-value mapping used to build HistoricJoaEndpoint params models.""" diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index 80f42bf..e7d0259 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -1,20 +1,16 @@ import pytest -from usajobsapi.endpoints.search import HiringPath, SearchEndpoint +from usajobsapi.endpoints.search import SearchEndpoint class TestSearchEndpointParams: - def test_to_params_serialization(self): - data = { - "keyword": "developer", - "location_names": ["City, ST", "Town, ST2"], - "radius": 25, - "relocation": True, - "job_category_codes": ["001", "002"], - "hiring_paths": [HiringPath.PUBLIC, HiringPath.VET], - } - params = SearchEndpoint.Params.model_validate(data) - assert params.to_params() == { + def test_to_params_serialization(self, search_params_kwargs): + """Validate Params.to_params uses USAJOBS aliases and formatting.""" + + params = SearchEndpoint.Params(**search_params_kwargs) + + serialized = params.to_params() + expected = { "Keyword": "developer", "LocationName": "City, ST;Town, ST2", "Radius": "25", @@ -23,6 +19,8 @@ def test_to_params_serialization(self): "HiringPath": "public;vet", } + assert serialized == expected + def test_radius_requires_location(self): with pytest.raises(ValueError): SearchEndpoint.Params.model_validate({"radius": 10}) diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index bd44566..88ddb3e 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -2,6 +2,7 @@ from __future__ import annotations +import datetime as dt from enum import StrEnum from typing import Annotated, Any, Dict, List, Optional @@ -17,7 +18,7 @@ ) from usajobsapi.endpoints._validators import isvalid_pay_grade -from usajobsapi.utils import _dump_by_alias +from usajobsapi.utils import _dump_by_alias, _normalize_date # Enums for query-params # --- @@ -530,13 +531,24 @@ class JobSummary(BaseModel): position_remuneration: List[SearchEndpoint.PositionRemuneration] = Field( default_factory=list, alias="PositionRemuneration" ) - publication_start_date: Optional[str] = Field( + publication_start_date: Optional[dt.date] = Field( default=None, alias="PublicationStartDate" ) - application_close_date: Optional[str] = Field( + application_close_date: Optional[dt.date] = Field( default=None, alias="ApplicationCloseDate" ) + @field_validator( + "publication_start_date", "application_close_date", mode="before" + ) + @classmethod + def _normalize_date_fields( + cls, value: None | dt.datetime | dt.date | str + ) -> Optional[dt.date]: + """Coerce date-like inputs to `datetime.date`.""" + + return _normalize_date(value) + def summary(self) -> Optional[str]: """Helper method returning the most descriptive summary for the job.""" From f8c10dd681d198da1082db8cb9b07cc3864337d9 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Mon, 29 Sep 2025 16:02:37 -0400 Subject: [PATCH 24/27] Rename JobSummary to JOAItem --- tests/unit/test_search.py | 2 +- usajobsapi/client.py | 4 ++-- usajobsapi/endpoints/search.py | 12 ++++++------ 3 files changed, 9 insertions(+), 9 deletions(-) diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index e7d0259..5b07faf 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -34,7 +34,7 @@ def test_remuneration_max_less_than_min(self): class TestSearchEndpointResponses: def test_job_summary_parses_nested_fields(self, job_summary_payload): - summary = SearchEndpoint.JobSummary.model_validate(job_summary_payload) + summary = SearchEndpoint.JOAItem.model_validate(job_summary_payload) assert summary.position_id == "24-123456" assert summary.position_uri == "https://example.com/job/1" diff --git a/usajobsapi/client.py b/usajobsapi/client.py index 49c2dd8..7f87907 100644 --- a/usajobsapi/client.py +++ b/usajobsapi/client.py @@ -184,11 +184,11 @@ def search_jobs_pages(self, **kwargs) -> Iterator[SearchEndpoint.Response]: current_page += 1 - def search_jobs_items(self, **kwargs) -> Iterator[SearchEndpoint.JobSummary]: + def search_jobs_items(self, **kwargs) -> Iterator[SearchEndpoint.JOAItem]: """Yield Job Search job items, handling pagination as needed. :yield: The job summary item. - :rtype: Iterator[SearchEndpoint.JobSummary] + :rtype: Iterator[SearchEndpoint.JOAItem] """ for resp in self.search_jobs_pages(**kwargs): for item in resp.jobs(): diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 88ddb3e..72c7e43 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -491,8 +491,8 @@ class UserArea(BaseModel): default=None, alias="Details" ) - class JobSummary(BaseModel): - """Normalized representation of a search result item.""" + class JOAItem(BaseModel): + """Represents a job opportunity annoucement object search result item.""" id: str = Field(alias="MatchedObjectId") position_id: Optional[str] = Field(default=None, alias="PositionID") @@ -606,14 +606,14 @@ class SearchResult(BaseModel): description="Array of job opportunity announcement objects that matched search criteria.", ) - def jobs(self) -> List[SearchEndpoint.JobSummary]: + def jobs(self) -> List[SearchEndpoint.JOAItem]: """Normalize the list of search results, skiping malformed payloads.""" - out: List[SearchEndpoint.JobSummary] = [] + out: List[SearchEndpoint.JOAItem] = [] for item in self.items: # Some responses nest the item under 'MatchedObjectDescriptor' descriptor = item.get("MatchedObjectDescriptor") or item try: - out.append(SearchEndpoint.JobSummary.model_validate(descriptor)) + out.append(SearchEndpoint.JOAItem.model_validate(descriptor)) except ValidationError: continue return out @@ -635,7 +635,7 @@ class Response(BaseModel): default=None, alias="SearchResult" ) - def jobs(self) -> List[SearchEndpoint.JobSummary]: + def jobs(self) -> List[SearchEndpoint.JOAItem]: """Helper method to directly expose parsed jobs in the response object.""" if not self.search_result: return [] From e83024a26e393303cb9436c0fe838d62a479e88d Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Mon, 29 Sep 2025 17:37:24 -0400 Subject: [PATCH 25/27] Correct JOAItem.id type --- tests/conftest.py | 2 +- tests/unit/test_client.py | 20 ++++++++++---------- tests/unit/test_search.py | 10 +++++----- usajobsapi/endpoints/search.py | 2 +- 4 files changed, 17 insertions(+), 17 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 5b334c1..1d41a0c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -25,7 +25,7 @@ def search_params_kwargs() -> Dict[str, Any]: def job_summary_payload(): """Sample payload matching the API's job summary schema.""" return { - "MatchedObjectId": "1", + "MatchedObjectId": 1, "PositionID": "24-123456", "PositionTitle": "Engineer", "OrganizationName": "NASA", diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index db2e082..fb5bfad 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -30,15 +30,15 @@ def test_search_jobs_pages_yields_pages(monkeypatch, search_result_item) -> None """Ensure search_jobs_pages iterates pages based on total counts.""" first_item = deepcopy(search_result_item) - first_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "1" + first_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 1 second_item = deepcopy(search_result_item) - second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "2" + second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 2 third_item = deepcopy(search_result_item) - third_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "3" + third_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 3 fourth_item = deepcopy(search_result_item) - fourth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "4" + fourth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 4 fifth_item = deepcopy(search_result_item) - fifth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "5" + fifth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 5 responses = [ SearchEndpoint.Response.model_validate( @@ -80,7 +80,7 @@ def test_search_jobs_pages_handles_missing_total( ) ) second_item = deepcopy(search_result_item) - second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = "3" + second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 3 second_page = SearchEndpoint.Response.model_validate( _build_search_payload([second_item], 1) ) @@ -152,11 +152,11 @@ def test_search_jobs_items_yields_jobs(monkeypatch, job_summary_payload) -> None client = USAJobsClient() first_payload = deepcopy(job_summary_payload) - first_payload["MatchedObjectId"] = "1" + first_payload["MatchedObjectId"] = 1 second_payload = deepcopy(job_summary_payload) - second_payload["MatchedObjectId"] = "2" + second_payload["MatchedObjectId"] = 2 third_payload = deepcopy(job_summary_payload) - third_payload["MatchedObjectId"] = "3" + third_payload["MatchedObjectId"] = 3 responses = [ _search_response_payload( @@ -185,7 +185,7 @@ def fake_search_jobs(self, **_): summaries = list(client.search_jobs_items(results_per_page=2)) - assert [summary.id for summary in summaries] == ["1", "2", "3"] + assert [summary.id for summary in summaries] == [1, 2, 3] # test historic_joa_pages diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index 5b07faf..132b80c 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -68,7 +68,7 @@ def test_job_summary_parses_nested_fields(self, job_summary_payload): def test_search_result_jobs_parsing(self, search_result_item): items = [ search_result_item, - {"MatchedObjectId": "2", "PositionTitle": "Analyst"}, + {"MatchedObjectId": 2, "PositionTitle": "Analyst"}, {"MatchedObjectDescriptor": {"MatchedObjectId": "3"}}, # invalid ] search_result = SearchEndpoint.SearchResult.model_validate( @@ -80,8 +80,8 @@ def test_search_result_jobs_parsing(self, search_result_item): ) jobs = search_result.jobs() assert len(jobs) == 2 - assert jobs[0].id == "1" - assert jobs[1].id == "2" + assert jobs[0].id == 1 + assert jobs[1].id == 2 def test_response_model_parsing(self, search_result_item): response = SearchEndpoint.Response.model_validate( @@ -103,7 +103,7 @@ def test_response_model_parsing(self, search_result_item): assert response.params.keyword == "python" assert response.search_result is not None jobs = response.search_result.jobs() - assert jobs[0].id == "1" + assert jobs[0].id == 1 def test_response_jobs_helper(self, search_result_item): response = SearchEndpoint.Response.model_validate( @@ -117,7 +117,7 @@ def test_response_jobs_helper(self, search_result_item): ) jobs = response.jobs() assert len(jobs) == 1 - assert jobs[0].id == "1" + assert jobs[0].id == 1 empty_response = SearchEndpoint.Response.model_validate({}) assert empty_response.jobs() == [] diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 72c7e43..284f5cb 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -494,7 +494,7 @@ class UserArea(BaseModel): class JOAItem(BaseModel): """Represents a job opportunity annoucement object search result item.""" - id: str = Field(alias="MatchedObjectId") + id: int = Field(alias="MatchedObjectId", description="Control Number") position_id: Optional[str] = Field(default=None, alias="PositionID") position_title: str = Field(alias="PositionTitle") position_uri: Optional[str] = Field(default=None, alias="PositionURI") From 718108c103070352b637aec2818cc25326efbae3 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Mon, 29 Sep 2025 22:06:59 -0400 Subject: [PATCH 26/27] Fix SearchResultItems members --- tests/conftest.py | 116 +++++++----- tests/unit/test_client.py | 20 +- tests/unit/test_search.py | 116 +++++------- usajobsapi/endpoints/search.py | 324 ++++++++++++++++++++++----------- 4 files changed, 348 insertions(+), 228 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 1d41a0c..05013df 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -22,60 +22,86 @@ def search_params_kwargs() -> Dict[str, Any]: @pytest.fixture -def job_summary_payload(): - """Sample payload matching the API's job summary schema.""" +def search_result_item(): return { "MatchedObjectId": 1, - "PositionID": "24-123456", - "PositionTitle": "Engineer", - "OrganizationName": "NASA", - "DepartmentName": "National Aeronautics and Space Administration", - "PositionURI": "https://example.com/job/1", - "ApplyURI": ["https://example.com/apply/1"], - "PositionLocationDisplay": "Houston, TX", - "PositionLocation": [ + "MatchedObjectDescriptor": { + "PositionID": "24-123456", + "PositionTitle": "Engineer", + "PositionURI": "https://example.com/job/1", + "ApplyURI": ["https://example.com/apply/1"], + "OrganizationName": "NASA", + "DepartmentName": "National Aeronautics and Space Administration", + "PositionLocationDisplay": "Houston, TX", + "PositionLocation": [ + { + "LocationName": "Houston, Texas", + "LocationCode": "TX1234", + "CountryCode": "US", + "CountrySubDivisionCode": "TX", + "CityName": "Houston", + "Latitude": "29.7604", + "Longitude": "-95.3698", + } + ], + "JobCategory": [{"Code": "0801", "Name": "General Engineering"}], + "JobGrade": [{"Code": "GS", "CurrentGrade": "12"}], + "PositionSchedule": [{"Code": "1", "Name": "Full-time"}], + "PositionOfferingType": [{"Code": "15317", "Name": "Permanent"}], + "MinimumRange": 50000.0, + "MaximumRange": 100000.0, + "PositionRemuneration": [ + { + "MinimumRange": "50000", + "MaximumRange": "100000", + "RateIntervalCode": "PA", + "Description": "Per Year", + } + ], + "ApplicationCloseDate": "2024-01-01", + "UserArea": { + "Details": { + "JobSummary": "Design and build spacecraft components.", + "WhoMayApply": { + "Name": "Open to the public", + "Code": "public", + }, + } + }, + }, + } + + +@pytest.fixture +def search_result_payload(search_result_item): + return { + "SearchResultCount": 3, + "SearchResultCountAll": 3, + "SearchResultItems": [ + search_result_item, { - "LocationName": "Houston, Texas", - "LocationCode": "TX1234", - "CountryCode": "US", - "CountrySubDivisionCode": "TX", - "CityName": "Houston", - "Latitude": "29.7604", - "Longitude": "-95.3698", - } - ], - "JobCategory": [{"Code": "0801", "Name": "General Engineering"}], - "JobGrade": [{"Code": "GS", "CurrentGrade": "12"}], - "PositionSchedule": [{"Code": "1", "Name": "Full-time"}], - "PositionOfferingType": [{"Code": "15317", "Name": "Permanent"}], - "MinimumRange": 50000.0, - "MaximumRange": 100000.0, - "PositionRemuneration": [ + "MatchedObjectId": 2, + "MatchedObjectDescriptor": {"PositionTitle": "Analyst"}, + }, { - "MinimumRange": "50000", - "MaximumRange": "100000", - "RateIntervalCode": "PA", - "Description": "Per Year", - } + "MatchedObjectId": 3, + "MatchedObjectDescriptor": {"PositionID": "3"}, + }, ], - "ApplicationCloseDate": "2024-01-01", - "UserArea": { - "Details": { - "JobSummary": "Design and build spacecraft components.", - "HiringPath": "public;vet", - "WhoMayApply": { - "Name": "Open to the public", - "Code": "public", - }, - } - }, } @pytest.fixture -def search_result_item(job_summary_payload): - """Wrap the job summary payload under the expected descriptor key.""" - return {"MatchedObjectDescriptor": job_summary_payload} +def search_response_payload(search_result_payload): + """Sample payload matching the API's job summary schema.""" + return { + "LanguageCode": "EN", + "SearchParameters": { + "Keyword": "python", + "LocationName": ["Atlanta,%20Georgia"], + }, + "SearchResult": search_result_payload, + } # historicjoa fixtures diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index fb5bfad..cd380f6 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -126,7 +126,7 @@ def fake_search(self, **_): def _search_response_payload( - items: list[dict], + items: list[dict | SearchEndpoint.JOAItem], count: int, total: int, page: int, @@ -135,8 +135,8 @@ def _search_response_payload( return { "LanguageCode": "EN", "SearchParameters": { - "page": page, - "results_per_page": results_per_page, + "Page": page, + "ResultsPerPage": results_per_page, }, "SearchResult": { "SearchResultCount": count, @@ -146,23 +146,23 @@ def _search_response_payload( } -def test_search_jobs_items_yields_jobs(monkeypatch, job_summary_payload) -> None: +def test_search_jobs_items_yields_jobs(monkeypatch, search_result_item) -> None: """Ensure search_jobs_items yields summaries across pages.""" client = USAJobsClient() - first_payload = deepcopy(job_summary_payload) + first_payload = deepcopy(search_result_item) first_payload["MatchedObjectId"] = 1 - second_payload = deepcopy(job_summary_payload) + second_payload = deepcopy(search_result_item) second_payload["MatchedObjectId"] = 2 - third_payload = deepcopy(job_summary_payload) + third_payload = deepcopy(search_result_item) third_payload["MatchedObjectId"] = 3 responses = [ _search_response_payload( [ - {"MatchedObjectDescriptor": first_payload}, - {"MatchedObjectDescriptor": second_payload}, + first_payload, + second_payload, ], 2, 3, @@ -170,7 +170,7 @@ def test_search_jobs_items_yields_jobs(monkeypatch, job_summary_payload) -> None 2, ), _search_response_payload( - [{"MatchedObjectDescriptor": third_payload}], + [third_payload], 1, 3, 2, diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py index 132b80c..9aa2e36 100644 --- a/tests/unit/test_search.py +++ b/tests/unit/test_search.py @@ -33,91 +33,65 @@ def test_remuneration_max_less_than_min(self): class TestSearchEndpointResponses: - def test_job_summary_parses_nested_fields(self, job_summary_payload): - summary = SearchEndpoint.JOAItem.model_validate(job_summary_payload) + def test_job_summary_parses_nested_fields(self, search_result_item): + summary = SearchEndpoint.JOAItem.model_validate(search_result_item) - assert summary.position_id == "24-123456" - assert summary.position_uri == "https://example.com/job/1" - assert summary.apply_uri == ["https://example.com/apply/1"] + assert summary.details.position_id == "24-123456" + assert summary.details.position_uri == "https://example.com/job/1" + assert summary.details.apply_uri == ["https://example.com/apply/1"] assert ( - summary.department_name == "National Aeronautics and Space Administration" + summary.details.department_name + == "National Aeronautics and Space Administration" ) - assert summary.locations_display == "Houston, TX" + assert summary.details.locations_display == "Houston, TX" - assert len(summary.locations) == 1 - location = summary.locations[0] + assert len(summary.details.locations) == 1 + location = summary.details.locations[0] assert location.city_name == "Houston" assert location.state_code == "TX" assert location.latitude == pytest.approx(29.7604) assert location.longitude == pytest.approx(-95.3698) - assert summary.job_categories[0].code == "0801" - assert summary.job_grades[0].current_grade == "12" - assert summary.position_schedules[0].name == "Full-time" - assert summary.position_offerings[0].code == "15317" - - assert summary.salary_range() == (50000.0, 100000.0) - assert summary.hiring_paths() == ["public", "vet"] - assert summary.summary() == "Design and build spacecraft components." - - assert summary.user_area - assert summary.user_area.details - assert summary.user_area.details.who_may_apply - assert summary.user_area.details.who_may_apply.name == "Open to the public" - - def test_search_result_jobs_parsing(self, search_result_item): - items = [ - search_result_item, - {"MatchedObjectId": 2, "PositionTitle": "Analyst"}, - {"MatchedObjectDescriptor": {"MatchedObjectId": "3"}}, # invalid - ] + assert summary.details.job_categories[0].code == "0801" + assert summary.details.job_grades[0].current_grade == "12" + assert summary.details.position_schedules[0].name == "Full-time" + assert summary.details.position_offerings[0].code == "15317" + + assert summary.details.summary() == "Design and build spacecraft components." + + assert summary.details.user_area + assert summary.details.user_area.details + assert summary.details.user_area.details.who_may_apply + assert ( + summary.details.user_area.details.who_may_apply.name == "Open to the public" + ) + + def test_search_result_jobs_parsing(self, search_result_payload): search_result = SearchEndpoint.SearchResult.model_validate( - { - "SearchResultCount": 3, - "SearchResultCountAll": 3, - "SearchResultItems": items, - } + search_result_payload ) - jobs = search_result.jobs() - assert len(jobs) == 2 + + jobs = search_result.items + assert len(jobs) == 3 assert jobs[0].id == 1 assert jobs[1].id == 2 + assert jobs[2].details.position_id == "3" - def test_response_model_parsing(self, search_result_item): - response = SearchEndpoint.Response.model_validate( - { - "LanguageCode": "EN", - "SearchParameters": { - "keyword": "python", - "location_names": ["Anywhere"], - }, - "SearchResult": { - "SearchResultCount": 1, - "SearchResultCountAll": 1, - "SearchResultItems": [search_result_item], - }, - } - ) - assert response.language == "EN" - assert response.params is not None - assert response.params.keyword == "python" - assert response.search_result is not None - jobs = response.search_result.jobs() - assert jobs[0].id == 1 + def test_response_model_parsing(self, search_response_payload): + resp = SearchEndpoint.Response.model_validate(search_response_payload) - def test_response_jobs_helper(self, search_result_item): - response = SearchEndpoint.Response.model_validate( - { - "SearchResult": { - "SearchResultCount": 1, - "SearchResultCountAll": 1, - "SearchResultItems": [search_result_item], - } - } - ) - jobs = response.jobs() - assert len(jobs) == 1 + assert resp.language == "EN" + assert resp.params is not None + assert resp.params.keyword == "python" + assert resp.search_result is not None + jobs = resp.search_result.items assert jobs[0].id == 1 - empty_response = SearchEndpoint.Response.model_validate({}) - assert empty_response.jobs() == [] + def test_response_jobs_helper(self, search_response_payload): + empty_resp = SearchEndpoint.Response.model_validate({}) + assert empty_resp.jobs() == [] + + resp = SearchEndpoint.Response.model_validate(search_response_payload) + jobs = resp.jobs() + assert len(jobs) == 3 + assert jobs[0].id == 1 diff --git a/usajobsapi/endpoints/search.py b/usajobsapi/endpoints/search.py index 284f5cb..966cabb 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -11,7 +11,6 @@ BaseModel, ConfigDict, Field, - ValidationError, ValidationInfo, field_validator, model_validator, @@ -104,13 +103,13 @@ class Params(BaseModel): keyword: Optional[str] = Field( None, - serialization_alias="Keyword", + alias="Keyword", description="Issues search to find hits based on a keyword. Optional. Keyword will search for all of the words specified (or synonyms of the word) throughout the job announcement.", examples=["https://data.usajobs.gov/api/search?Keyword=Software"], ) position_title: Optional[str] = Field( None, - serialization_alias="PositionTitle", + alias="PositionTitle", description=""" Issues search to find hits in the title of the job. @@ -123,7 +122,7 @@ class Params(BaseModel): remuneration_min: Optional[int] = Field( None, - serialization_alias="RemunerationMinimumAmount", + alias="RemunerationMinimumAmount", description=""" Issues search to find hits with the minimum salary specified. @@ -135,7 +134,7 @@ class Params(BaseModel): ) remuneration_max: Optional[int] = Field( None, - serialization_alias="RemunerationMaximumAmount", + alias="RemunerationMaximumAmount", description=""" Issues search to find hits with the maximum salary specified. @@ -150,7 +149,7 @@ class Params(BaseModel): pay_grade_high: Annotated[Optional[str], AfterValidator(isvalid_pay_grade)] = ( Field( None, - serialization_alias="PayGradeHigh", + alias="PayGradeHigh", description=""" Issues search to find hits with the maximum pay grade specified. Must be 01 through 15. This is the ending grade for the job. (Caution: Fed speak ahead but it cannot be helped.) The grade along with series is used by the Federal government to categorize and define jobs. @@ -168,7 +167,7 @@ class Params(BaseModel): pay_grade_low: Annotated[Optional[str], AfterValidator(isvalid_pay_grade)] = ( Field( None, - serialization_alias="PayGradeLow", + alias="PayGradeLow", description="Issues search to find hits with the minimum pay grade specified. Must be 01 through 15. This is the beginning grade for the job. See PayGradeHigh for more information.", examples=[ "https://data.usajobs.gov/api/Search?PayGradeLow=04", @@ -179,32 +178,32 @@ class Params(BaseModel): job_category_codes: List[str] = Field( default_factory=list, - serialization_alias="JobCategoryCode", + alias="JobCategoryCode", description="Issues a search to find hits with the job category series specified.", examples=["https://data.usajobs.gov/api/Search?JobCategoryCode=0830"], ) position_schedule_type_codes: List[int] = Field( default_factory=list, - serialization_alias="PositionScheduleTypeCode", + alias="PositionScheduleTypeCode", description="Issues a search to find hits for jobs matching the specified job schedule. This field is also known as work schedule.", examples=["https://data.usajobs.gov/api/Search?PositionSchedule=4"], ) position_offering_type_codes: List[int] = Field( default_factory=list, - serialization_alias="PositionOfferingTypeCode", + alias="PositionOfferingTypeCode", description="Issues a search to find jobs within the specified type. This field is also known as Work Type.", ) organization: List[str] = Field( default_factory=list, - serialization_alias="Organization", + alias="Organization", description="Issues a search to find jobs for the specified agency using the Agency Subelement Code.", examples=["https://data.usajobs.gov/api/Search?Organization=TR"], ) location_names: List[str] = Field( default_factory=list, - serialization_alias="LocationName", + alias="LocationName", description="Issues a search to find hits within the specified location. This is the city or military installation name. LocationName simplifies location based search as the user does not need to know or account for each and every Location Code. LocationName will search for all location codes and ZIP codes that have that specific description.", examples=[ "https://data.usajobs.gov/api/Search?LocationName=Washington%20DC,%20District%20of%20Columbia", @@ -214,7 +213,7 @@ class Params(BaseModel): travel_percentage: List[int] = Field( default_factory=list, - serialization_alias="TravelPercentage", + alias="TravelPercentage", description="Issues a search to find hits for jobs matching the specified travel level.", examples=[ "https://data.usajobs.gov/api/Search?TravelPercentage=0", @@ -225,12 +224,12 @@ class Params(BaseModel): ) relocation: Optional[bool] = Field( None, - serialization_alias="RelocationIndicator", + alias="RelocationIndicator", description="Issues a search to find hits for jobs matching the relocation filter.", ) security_clearance_required: List[int] = Field( default_factory=list, - serialization_alias="SecurityClearanceRequired", + alias="SecurityClearanceRequired", description="Issues a search to find hits for jobs matching the specified security clearance.", examples=[ "https://data.usajobs.gov/api/Search?SecurityClearanceRequired=1" @@ -241,14 +240,14 @@ class Params(BaseModel): supervisory_status: Optional[str] = Field( None, - serialization_alias="SupervisoryStatus", + alias="SupervisoryStatus", description="Issues a search to find hits for jobs matching the specified supervisory status.", ) days_since_posted: Annotated[ Optional[int], Field( - serialization_alias="DatePosted", + alias="DatePosted", description="Issues a search to find hits for jobs that were posted within the number of days specified.", strict=True, ge=0, @@ -257,13 +256,13 @@ class Params(BaseModel): ] = None job_grade_codes: List[str] = Field( default_factory=list, - serialization_alias="JobGradeCode", + alias="JobGradeCode", description="Issues a search to find hits for jobs matching the grade code specified. This field is also known as Pay Plan.", ) sort_field: Optional[SortField] = Field( None, - serialization_alias="SortField", + alias="SortField", description="Issues a search that will be sorted by the specified field.", examples=[ "https://data.usajobs.gov/api/Search?PositionTitle=Electrical&SortField=PositionTitle" @@ -271,14 +270,14 @@ class Params(BaseModel): ) sort_direction: Optional[SortDirection] = Field( None, - serialization_alias="SortDirection", + alias="SortDirection", description="Issues a search that will be sorted by the SortField specified, in the direction specified.", ) page: Annotated[ Optional[int], Field( - serialization_alias="Page", + alias="Page", description="Issues a search to pull the paged results specified.", strict=True, ge=1, @@ -287,7 +286,7 @@ class Params(BaseModel): results_per_page: Annotated[ Optional[int], Field( - serialization_alias="ResultsPerPage", + alias="ResultsPerPage", description="Issues a search and returns the page size specified. In this example, 25 jobs will be return for the first page.", strict=True, ge=1, @@ -297,7 +296,7 @@ class Params(BaseModel): who_may_apply: Optional[WhoMayApply] = Field( None, - serialization_alias="WhoMayApply", + alias="WhoMayApply", description="Issues a search to find hits based on the desired candidate designation. In this case, public will find jobs that U.S. citizens can apply for.", examples=["https://data.usajobs.gov/api/Search?WhoMayApply=public"], ) @@ -305,7 +304,7 @@ class Params(BaseModel): radius: Annotated[ Optional[int], Field( - serialization_alias="Radius", + alias="Radius", description="Issues a search when used along with LocationName, will expand the locations, based on the radius specified.", examples=[ "https://data.usajobs.gov/api/Search?LocationName=Norfolk%20Virginia&Radius=75" @@ -316,7 +315,7 @@ class Params(BaseModel): ] = None fields: Optional[FieldsMinMax] = Field( None, - serialization_alias="Fields", + alias="Fields", description="Issues a search that will return the minimum fields or maximum number of fields in the job. Min returns only the job summary.", examples=[ "https://data.usajobs.gov/api/Search?TravelPercentage=7&Fields=full", @@ -326,25 +325,25 @@ class Params(BaseModel): salary_bucket: List[int] = Field( default_factory=list, - serialization_alias="SalaryBucket", + alias="SalaryBucket", description="Issues a search that will find hits for salaries matching the grouping specified. Buckets are assigned based on salary ranges.", ) grade_bucket: List[int] = Field( default_factory=list, - serialization_alias="GradeBucket", + alias="GradeBucket", description="Issues a search that will find hits for grades that match the grouping specified.", ) hiring_paths: List[HiringPath] = Field( default_factory=list, - serialization_alias="HiringPath", + alias="HiringPath", description="Issues a search that will find hits for hiring paths that match the hiring paths specified.", examples=["https://data.usajobs.gov/api/Search?HiringPath=public"], ) mission_critical_tags: List[str] = Field( default_factory=list, - serialization_alias="MissionCriticalTags", + alias="MissionCriticalTags", description="Issues a search that will find hits for mission critical tags that match the grouping specified.", examples=[ "https://data.usajobs.gov/api/Search?MissionCriticalTags=STEM&Fields=full" @@ -353,7 +352,7 @@ class Params(BaseModel): position_sensitivity: List[int] = Field( default_factory=list, - serialization_alias="PositionSensitivity", + alias="PositionSensitivity", description="Issues a search that will find hits for jobs matching the position sensitivity and risk specified.", examples=["https://data.usajobs.gov/api/Search?PositionSensitivity=1"], ge=1, @@ -362,7 +361,7 @@ class Params(BaseModel): remote_indicator: Optional[bool] = Field( None, - serialization_alias="RemoteIndicator", + alias="RemoteIndicator", description="Issues a search to find hits for jobs matching the remote filter.", ) @@ -475,13 +474,103 @@ class WhoMayApplyInfo(BaseModel): name: Optional[str] = Field(default=None, alias="Name") code: Optional[str] = Field(default=None, alias="Code") + class PositionFormatDesc(BaseModel): + """Represents a quick summary of the job opportunity announcement.""" + + content: Optional[str] = Field(default=None, alias="Content") + label: Optional[str] = Field(default=None, alias="Label") + label_desc: Optional[str] = Field(default=None, alias="LabelDescription") + class UserAreaDetails(BaseModel): """Represents metadata stored under the UserArea.Details field.""" - job_summary: Optional[str] = Field(default=None, alias="JobSummary") - hiring_path: Optional[str] = Field(default=None, alias="HiringPath") + duties: Optional[str] = Field( + default=None, + alias="MajorDuties", + description="Description of the duties of the job.", + ) + education: Optional[str] = Field( + default=None, + alias="Education", + description="Education required or preferred by applicants.", + ) + requirements: Optional[str] = Field( + default=None, + alias="Requirements", + description="Key Requirements of the job opportunity.", + ) + evaluations: Optional[str] = Field( + default=None, + alias="Evaluations", + description="Qualification requirements of the job opportunity.", + ) + how_apply: Optional[str] = Field( + default=None, + alias="HowToApply", + description="Description of the steps to take to apply for the job opportunity.", + ) + what_next: Optional[str] = Field( + default=None, + alias="WhatToExpectNext", + description="Description of what can be expected during the application process.", + ) + req_docs: Optional[str] = Field( + default=None, + alias="RequiredDocuments", + description="Required documents when applying for the job opportunity.", + ) + benefits: Optional[str] = Field( + default=None, + alias="Benefits being offered as part of the job opportunity.", + description="BenefitsBenefits", + ) + benefits_url: Optional[str] = Field( + default=None, + alias="BenefitsUrl", + description="URL to view benefit details being offered.", + ) + other_info: Optional[str] = Field( + default=None, + alias="OtherInformation", + description="Additional information about the job opportunity.", + ) + key_reqs: List[str] = Field( + default_factory=list, + alias="KeyRequirements", + description="List of requirements for the job opportunity.", + ) + job_summary: Optional[str] = Field( + default=None, + alias="JobSummary", + description="Summary of the job opportunity.", + ) + who_may_apply: Optional[SearchEndpoint.WhoMayApplyInfo] = Field( - default=None, alias="WhoMayApply" + default=None, + alias="WhoMayApply", + description="Object that contains values for name and code of who may apply to the job opportunity.", + ) + + low_grade: Optional[str] = Field( + default=None, + alias="LowGrade", + description="Lowest potential grade level for the job opportunity.", + ) + high_grade: Optional[str] = Field( + default=None, + alias="HighGrade", + description="Highest potential grade level for the job opportunity.", + ) + + sub_agency: Optional[str] = Field( + default=None, + alias="SubAgencyName", + description="SubAgencyName", + ) + org_codes: Optional[str] = Field( + default=None, + alias="OrganizationCodes", + description="Organization codes separated by a slash (/).", ) class UserArea(BaseModel): @@ -490,56 +579,121 @@ class UserArea(BaseModel): details: Optional[SearchEndpoint.UserAreaDetails] = Field( default=None, alias="Details" ) + is_radial_search: Optional[bool] = Field( + default=None, + alias="IsRadialSearch", + description="Was a radial search preformed.", + ) - class JOAItem(BaseModel): - """Represents a job opportunity annoucement object search result item.""" + class JOADescriptor(BaseModel): + position_id: Optional[str] = Field( + default=None, alias="PositionID", description="Job Announcement Number" + ) + position_title: Optional[str] = Field( + default=None, + alias="PositionTitle", + description="Title of the job offering.", + ) + position_uri: Optional[str] = Field( + default=None, + alias="PositionURI", + description="URI to view the job offering.", + ) + apply_uri: List[str] = Field( + default_factory=list, + alias="ApplyURI", + description="URI to apply for the job offering.", + ) - id: int = Field(alias="MatchedObjectId", description="Control Number") - position_id: Optional[str] = Field(default=None, alias="PositionID") - position_title: str = Field(alias="PositionTitle") - position_uri: Optional[str] = Field(default=None, alias="PositionURI") - # URI to apply for the job offering - apply_uri: List[str] = Field(default_factory=list, alias="ApplyURI") - organization_name: Optional[str] = Field(default=None, alias="OrganizationName") - department_name: Optional[str] = Field(default=None, alias="DepartmentName") locations_display: Optional[str] = Field( default=None, alias="PositionLocationDisplay" ) locations: List[SearchEndpoint.JobLocation] = Field( - default_factory=list, alias="PositionLocation" + default_factory=list, + alias="PositionLocation", + description="Contains values for location name, country, country subdivision, city, latitude and longitude.", + ) + + organization_name: Optional[str] = Field( + default=None, + alias="OrganizationName", + description="Name of the organization or agency offering the position.", + ) + department_name: Optional[str] = Field( + default=None, + alias="DepartmentName", + description="Name of the department within the organization or agency offering the position.", ) + job_categories: List[SearchEndpoint.JobCategory] = Field( - default_factory=list, alias="JobCategory" + default_factory=list, + alias="JobCategory", + description="List of job category objects that contain values for name and code.", ) job_grades: List[SearchEndpoint.JobGrade] = Field( - default_factory=list, alias="JobGrade" + default_factory=list, + alias="JobGrade", + description="List of job grade objects that contains an code value. This field is also known as Pay Plan.", ) position_schedules: List[SearchEndpoint.PositionSchedule] = Field( - default_factory=list, alias="PositionSchedule" + default_factory=list, + alias="PositionSchedule", + description="List of position schedule objects that contains values for name and code.", ) position_offerings: List[SearchEndpoint.PositionOfferingType] = Field( - default_factory=list, alias="PositionOfferingType" - ) - user_area: Optional[SearchEndpoint.UserArea] = Field( - default=None, alias="UserArea" + default_factory=list, + alias="PositionOfferingType", + description="List of position offering type objects that contains values for name and code. See PositionOfferingType in paramters above for list of code values.", ) + qualification_summary: Optional[str] = Field( - default=None, alias="QualificationSummary" + default=None, + alias="QualificationSummary", + description="Summary of qualifications for the job offering.", ) - min_salary: Optional[float] = Field(default=None, alias="MinimumRange") - max_salary: Optional[float] = Field(default=None, alias="MaximumRange") + position_remuneration: List[SearchEndpoint.PositionRemuneration] = Field( - default_factory=list, alias="PositionRemuneration" + default_factory=list, + alias="PositionRemuneration", + description="List of position remuneration objects that contains MinimumRange, MaximumRange and RateIntervalCode. Gives the pay range and frequency.", + ) + position_start_date: Optional[dt.date] = Field( + default=None, + alias="PositionStartDate", + description="The date the job opportunity will be open to applications.", + ) + position_end_date: Optional[dt.date] = Field( + default=None, + alias="PositionEndDate", + description="Last date the job opportunity will be posted", ) publication_start_date: Optional[dt.date] = Field( - default=None, alias="PublicationStartDate" + default=None, + alias="PublicationStartDate", + description="Date the job opportunity is posted", ) application_close_date: Optional[dt.date] = Field( - default=None, alias="ApplicationCloseDate" + default=None, + alias="ApplicationCloseDate", + description="Last date applications will be accepted for the job opportunity", + ) + + position_format_desc: List[SearchEndpoint.PositionFormatDesc] = Field( + default_factory=list, + alias="PositionFormattedDescription", + description="Provides quick summary of job opportunity.", + ) + + user_area: Optional[SearchEndpoint.UserArea] = Field( + default=None, alias="UserArea" ) @field_validator( - "publication_start_date", "application_close_date", mode="before" + "position_start_date", + "position_end_date", + "publication_start_date", + "application_close_date", + mode="before", ) @classmethod def _normalize_date_fields( @@ -554,38 +708,16 @@ def summary(self) -> Optional[str]: if self.user_area and self.user_area.details: details = self.user_area.details - if details and details.job_summary: + if self.user_area.details and details.job_summary: return details.job_summary return self.qualification_summary - def salary_range(self) -> tuple[Optional[float], Optional[float]]: - """Helper method returning the salary range inferred from remuneration metadata.""" - - if self.position_remuneration: - minima = [ - r.minimum - for r in self.position_remuneration - if r.minimum is not None - ] - maxima = [ - r.maximum - for r in self.position_remuneration - if r.maximum is not None - ] - min_val = min(minima) if minima else None - max_val = max(maxima) if maxima else None - if min_val is not None or max_val is not None: - return min_val, max_val - return self.min_salary, self.max_salary - - def hiring_paths(self) -> List[str]: - """Helper method returning normalized hiring path codes from the user area.""" + class JOAItem(BaseModel): + """Represents a job opportunity annoucement object search result item.""" - if self.user_area and self.user_area.details: - raw = self.user_area.details.hiring_path - if raw: - return [path.strip() for path in raw.split(";") if path.strip()] - return [] + id: int = Field(alias="MatchedObjectId", description="Control Number") + details: SearchEndpoint.JOADescriptor = Field(alias="MatchedObjectDescriptor") + rank: Optional[float] = Field(default=None, alias="RelevanceRank") class SearchResult(BaseModel): """Model of paginated search results.""" @@ -600,24 +732,12 @@ class SearchResult(BaseModel): alias="SearchResultCountAll", description="Total Number of records that matched search criteria.", ) - items: List[Dict[str, Any]] = Field( + items: List[SearchEndpoint.JOAItem] = Field( default_factory=list, alias="SearchResultItems", description="Array of job opportunity announcement objects that matched search criteria.", ) - def jobs(self) -> List[SearchEndpoint.JOAItem]: - """Normalize the list of search results, skiping malformed payloads.""" - out: List[SearchEndpoint.JOAItem] = [] - for item in self.items: - # Some responses nest the item under 'MatchedObjectDescriptor' - descriptor = item.get("MatchedObjectDescriptor") or item - try: - out.append(SearchEndpoint.JOAItem.model_validate(descriptor)) - except ValidationError: - continue - return out - class Response(BaseModel): """Declarative definition of the endpoint's response object.""" @@ -636,7 +756,7 @@ class Response(BaseModel): ) def jobs(self) -> List[SearchEndpoint.JOAItem]: - """Helper method to directly expose parsed jobs in the response object.""" + """Helper method to directly expose search result items from the response object.""" if not self.search_result: return [] - return self.search_result.jobs() + return self.search_result.items From 66c22410ebf272ed0c04b2e3a6e455b5c311b534 Mon Sep 17 00:00:00 2001 From: Patrick Cox Date: Mon, 29 Sep 2025 22:21:48 -0400 Subject: [PATCH 27/27] Clean-up unit tests and fixtures --- tests/conftest.py | 524 ++++++++++++++++++++++++-------------- tests/unit/test_client.py | 287 +++++++++------------ 2 files changed, 445 insertions(+), 366 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 05013df..049accb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,16 +1,163 @@ -from typing import Any, Dict, List +from collections.abc import Callable, Sequence +from copy import deepcopy +from typing import Any import pytest from usajobsapi.endpoints.search import HiringPath -# search fixtures -# --- +_DEFAULT_SEARCH_RESULT_ITEM: dict[str, Any] = { + "MatchedObjectId": 1, + "MatchedObjectDescriptor": { + "PositionID": "24-123456", + "PositionTitle": "Engineer", + "PositionURI": "https://example.com/job/1", + "ApplyURI": ["https://example.com/apply/1"], + "OrganizationName": "NASA", + "DepartmentName": "National Aeronautics and Space Administration", + "PositionLocationDisplay": "Houston, TX", + "PositionLocation": [ + { + "LocationName": "Houston, Texas", + "LocationCode": "TX1234", + "CountryCode": "US", + "CountrySubDivisionCode": "TX", + "CityName": "Houston", + "Latitude": "29.7604", + "Longitude": "-95.3698", + } + ], + "JobCategory": [{"Code": "0801", "Name": "General Engineering"}], + "JobGrade": [{"Code": "GS", "CurrentGrade": "12"}], + "PositionSchedule": [{"Code": "1", "Name": "Full-time"}], + "PositionOfferingType": [{"Code": "15317", "Name": "Permanent"}], + "MinimumRange": 50000.0, + "MaximumRange": 100000.0, + "PositionRemuneration": [ + { + "MinimumRange": "50000", + "MaximumRange": "100000", + "RateIntervalCode": "PA", + "Description": "Per Year", + } + ], + "ApplicationCloseDate": "2024-01-01", + "UserArea": { + "Details": { + "JobSummary": "Design and build spacecraft components.", + "WhoMayApply": { + "Name": "Open to the public", + "Code": "public", + }, + } + }, + }, +} + +_DEFAULT_HISTORICJOA_ITEMS: list[dict[str, Any]] = [ + { + "usajobsControlNumber": 123456789, + "hiringAgencyCode": "NASA", + "hiringAgencyName": "National Aeronautics and Space Administration", + "hiringDepartmentCode": "NAT", + "hiringDepartmentName": "Department of Science", + "agencyLevel": 2, + "agencyLevelSort": "Department of Science\\NASA", + "appointmentType": "Permanent", + "workSchedule": "Full-time", + "payScale": "GS", + "salaryType": "Per Year", + "vendor": "USASTAFFING", + "travelRequirement": "Occasional travel", + "teleworkEligible": "Y", + "serviceType": "Competitive", + "securityClearanceRequired": "Y", + "securityClearance": "Secret", + "whoMayApply": "United States Citizens", + "announcementClosingTypeCode": "C", + "announcementClosingTypeDescription": "Closing Date", + "positionOpenDate": "2020-01-01", + "positionCloseDate": "2020-02-01", + "positionExpireDate": None, + "announcementNumber": "NASA-20-001", + "hiringSubelementName": "Space Operations", + "positionTitle": "Data Scientist", + "minimumGrade": "12", + "maximumGrade": "13", + "promotionPotential": "13", + "minimumSalary": 90000.0, + "maximumSalary": 120000.0, + "supervisoryStatus": "N", + "drugTestRequired": "N", + "relocationExpensesReimbursed": "Y", + "totalOpenings": "3", + "disableApplyOnline": "N", + "positionOpeningStatus": "Accepting Applications", + "hiringPaths": [{"hiringPath": "The public"}], + "jobCategories": [{"series": "1550"}], + "positionLocations": [ + { + "positionLocationCity": "Houston", + "positionLocationState": "Texas", + "positionLocationCountry": "United States", + } + ], + }, + { + "usajobsControlNumber": 987654321, + "hiringAgencyCode": "DOE", + "hiringAgencyName": "Department of Energy", + "hiringDepartmentCode": "ENG", + "hiringDepartmentName": "Department of Energy", + "agencyLevel": 1, + "agencyLevelSort": "Department of Energy", + "appointmentType": "Term", + "workSchedule": "Part-time", + "payScale": "GS", + "salaryType": "Per Year", + "vendor": "OTHER", + "travelRequirement": "Not required", + "teleworkEligible": "N", + "serviceType": None, + "securityClearanceRequired": "N", + "securityClearance": "Not Required", + "whoMayApply": "Agency Employees Only", + "announcementClosingTypeCode": None, + "announcementClosingTypeDescription": None, + "positionOpenDate": "2020-03-01", + "positionCloseDate": "2020-04-01", + "positionExpireDate": "2020-04-15", + "announcementNumber": "DOE-20-ENG", + "hiringSubelementName": "Energy Research", + "positionTitle": "Backend Engineer", + "minimumGrade": "11", + "maximumGrade": "12", + "promotionPotential": None, + "minimumSalary": 80000.0, + "maximumSalary": 110000.0, + "supervisoryStatus": "Y", + "drugTestRequired": "Y", + "relocationExpensesReimbursed": "N", + "totalOpenings": "1", + "disableApplyOnline": "Y", + "positionOpeningStatus": "Closed", + "hiringPaths": [{"hiringPath": "Government employees"}], + "jobCategories": [{"series": "2210"}], + "positionLocations": [ + { + "positionLocationCity": "Washington", + "positionLocationState": "District of Columbia", + "positionLocationCountry": "United States", + } + ], + }, +] @pytest.fixture -def search_params_kwargs() -> Dict[str, Any]: - """Field-value mapping used to build SearchEndpoint params models.""" +def search_params_kwargs() -> dict[str, Any]: + """Field-value mapping used to build ``SearchEndpoint.Params`` models.""" + return { "keyword": "developer", "location_names": ["City, ST", "Town, ST2"], @@ -22,95 +169,143 @@ def search_params_kwargs() -> Dict[str, Any]: @pytest.fixture -def search_result_item(): - return { - "MatchedObjectId": 1, - "MatchedObjectDescriptor": { - "PositionID": "24-123456", - "PositionTitle": "Engineer", - "PositionURI": "https://example.com/job/1", - "ApplyURI": ["https://example.com/apply/1"], - "OrganizationName": "NASA", - "DepartmentName": "National Aeronautics and Space Administration", - "PositionLocationDisplay": "Houston, TX", - "PositionLocation": [ - { - "LocationName": "Houston, Texas", - "LocationCode": "TX1234", - "CountryCode": "US", - "CountrySubDivisionCode": "TX", - "CityName": "Houston", - "Latitude": "29.7604", - "Longitude": "-95.3698", - } - ], - "JobCategory": [{"Code": "0801", "Name": "General Engineering"}], - "JobGrade": [{"Code": "GS", "CurrentGrade": "12"}], - "PositionSchedule": [{"Code": "1", "Name": "Full-time"}], - "PositionOfferingType": [{"Code": "15317", "Name": "Permanent"}], - "MinimumRange": 50000.0, - "MaximumRange": 100000.0, - "PositionRemuneration": [ - { - "MinimumRange": "50000", - "MaximumRange": "100000", - "RateIntervalCode": "PA", - "Description": "Per Year", - } - ], - "ApplicationCloseDate": "2024-01-01", - "UserArea": { - "Details": { - "JobSummary": "Design and build spacecraft components.", - "WhoMayApply": { - "Name": "Open to the public", - "Code": "public", - }, - } - }, - }, - } +def make_search_result_item() -> Callable[..., dict[str, Any]]: + """Return a factory that produces search result item payloads.""" + + def _make( + *, + matched_object_id: int = 1, + descriptor_overrides: dict[str, Any] | None = None, + **overrides: Any, + ) -> dict[str, Any]: + item = deepcopy(_DEFAULT_SEARCH_RESULT_ITEM) + item["MatchedObjectId"] = matched_object_id + descriptor = item["MatchedObjectDescriptor"] + if descriptor_overrides: + descriptor.update(descriptor_overrides) + if overrides: + item.update(overrides) + return item + + return _make @pytest.fixture -def search_result_payload(search_result_item): - return { - "SearchResultCount": 3, - "SearchResultCountAll": 3, - "SearchResultItems": [ - search_result_item, - { - "MatchedObjectId": 2, - "MatchedObjectDescriptor": {"PositionTitle": "Analyst"}, - }, - { - "MatchedObjectId": 3, - "MatchedObjectDescriptor": {"PositionID": "3"}, - }, - ], - } +def search_result_item( + make_search_result_item: Callable[..., dict[str, Any]], +) -> dict[str, Any]: + """Default sample search result item used across unit tests.""" + + return make_search_result_item() @pytest.fixture -def search_response_payload(search_result_payload): - """Sample payload matching the API's job summary schema.""" - return { - "LanguageCode": "EN", - "SearchParameters": { - "Keyword": "python", - "LocationName": ["Atlanta,%20Georgia"], - }, - "SearchResult": search_result_payload, - } +def make_search_result_payload( + make_search_result_item: Callable[..., dict[str, Any]], +) -> Callable[..., dict[str, Any]]: + """Return a factory that produces serialized search result payloads.""" + + def _make( + *, + items: Sequence[dict[str, Any]] | None = None, + count: int | None = None, + total: int | None = None, + include_total: bool = True, + ) -> dict[str, Any]: + payload_items = ( + [deepcopy(item) for item in items] + if items is not None + else [ + make_search_result_item(matched_object_id=1), + make_search_result_item( + matched_object_id=2, + descriptor_overrides={"PositionTitle": "Analyst"}, + ), + make_search_result_item( + matched_object_id=3, + descriptor_overrides={"PositionID": "3"}, + ), + ] + ) + result_count = count if count is not None else len(payload_items) + payload: dict[str, Any] = { + "SearchResultCount": result_count, + "SearchResultItems": payload_items, + } + if include_total: + total_count = total if total is not None else result_count + payload["SearchResultCountAll"] = total_count + return payload + return _make -# historicjoa fixtures -# --- + +@pytest.fixture +def search_result_payload( + make_search_result_payload: Callable[..., dict[str, Any]], +) -> dict[str, Any]: + """Serialized payload matching the API's job summary schema.""" + + return make_search_result_payload() + + +@pytest.fixture +def make_search_response_payload( + make_search_result_payload: Callable[..., dict[str, Any]], +) -> Callable[..., dict[str, Any]]: + """Return a factory that produces serialized search endpoint responses.""" + + def _make( + *, + items: Sequence[dict[str, Any]] | None = None, + count: int | None = None, + total: int | None = None, + include_total: bool = True, + language: str = "EN", + keyword: str | None = "python", + location_names: Sequence[str] | None = ("Atlanta,%20Georgia",), + page: int | None = None, + results_per_page: int | None = None, + extra_params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + search_params: dict[str, Any] = {} + if keyword is not None: + search_params["Keyword"] = keyword + if location_names: + search_params["LocationName"] = list(location_names) + if page is not None: + search_params["Page"] = page + if results_per_page is not None: + search_params["ResultsPerPage"] = results_per_page + if extra_params: + search_params.update(extra_params) + + return { + "LanguageCode": language, + "SearchParameters": search_params, + "SearchResult": make_search_result_payload( + items=items, + count=count, + total=total, + include_total=include_total, + ), + } + + return _make + + +@pytest.fixture +def search_response_payload( + make_search_response_payload: Callable[..., dict[str, Any]], +) -> dict[str, Any]: + """Serialized search endpoint response payload.""" + + return make_search_response_payload() @pytest.fixture -def historicjoa_params_kwargs() -> Dict[str, str]: - """Field-value mapping used to build HistoricJoaEndpoint params models.""" +def historicjoa_params_kwargs() -> dict[str, str]: + """Field-value mapping used to build ``HistoricJoaEndpoint.Params`` models.""" return { "hiring_agency_codes": "AGENCY1", @@ -127,118 +322,57 @@ def historicjoa_params_kwargs() -> Dict[str, str]: @pytest.fixture -def historicjoa_response_payload() -> Dict[str, object]: - """Serialized Historic JOA response payload mimicking the USAJOBS API.""" +def make_historicjoa_item() -> Callable[..., dict[str, Any]]: + """Return a factory that produces historic JOA item payloads.""" - return { - "paging": { - "metadata": { - "totalCount": 2, - "pageSize": 2, - "continuationToken": "NEXTTOKEN", - }, - "next": "https://example.invalid/historicjoa?page=2", - }, - "data": _historicjoa_items(), - } + def _make(*, base: int = 0, **overrides: Any) -> dict[str, Any]: + item = deepcopy(_DEFAULT_HISTORICJOA_ITEMS[base]) + if overrides: + item.update(overrides) + return item + return _make -def _historicjoa_items() -> List[Dict[str, object]]: - return [ - { - "usajobsControlNumber": 123456789, - "hiringAgencyCode": "NASA", - "hiringAgencyName": "National Aeronautics and Space Administration", - "hiringDepartmentCode": "NAT", - "hiringDepartmentName": "Department of Science", - "agencyLevel": 2, - "agencyLevelSort": "Department of Science\\NASA", - "appointmentType": "Permanent", - "workSchedule": "Full-time", - "payScale": "GS", - "salaryType": "Per Year", - "vendor": "USASTAFFING", - "travelRequirement": "Occasional travel", - "teleworkEligible": "Y", - "serviceType": "Competitive", - "securityClearanceRequired": "Y", - "securityClearance": "Secret", - "whoMayApply": "United States Citizens", - "announcementClosingTypeCode": "C", - "announcementClosingTypeDescription": "Closing Date", - "positionOpenDate": "2020-01-01", - "positionCloseDate": "2020-02-01", - "positionExpireDate": None, - "announcementNumber": "NASA-20-001", - "hiringSubelementName": "Space Operations", - "positionTitle": "Data Scientist", - "minimumGrade": "12", - "maximumGrade": "13", - "promotionPotential": "13", - "minimumSalary": 90000.0, - "maximumSalary": 120000.0, - "supervisoryStatus": "N", - "drugTestRequired": "N", - "relocationExpensesReimbursed": "Y", - "totalOpenings": "3", - "disableApplyOnline": "N", - "positionOpeningStatus": "Accepting Applications", - "hiringPaths": [{"hiringPath": "The public"}], - "jobCategories": [{"series": "1550"}], - "positionLocations": [ - { - "positionLocationCity": "Houston", - "positionLocationState": "Texas", - "positionLocationCountry": "United States", - } - ], - }, - { - "usajobsControlNumber": 987654321, - "hiringAgencyCode": "DOE", - "hiringAgencyName": "Department of Energy", - "hiringDepartmentCode": "ENG", - "hiringDepartmentName": "Department of Energy", - "agencyLevel": 1, - "agencyLevelSort": "Department of Energy", - "appointmentType": "Term", - "workSchedule": "Part-time", - "payScale": "GS", - "salaryType": "Per Year", - "vendor": "OTHER", - "travelRequirement": "Not required", - "teleworkEligible": "N", - "serviceType": None, - "securityClearanceRequired": "N", - "securityClearance": "Not Required", - "whoMayApply": "Agency Employees Only", - "announcementClosingTypeCode": None, - "announcementClosingTypeDescription": None, - "positionOpenDate": "2020-03-01", - "positionCloseDate": "2020-04-01", - "positionExpireDate": "2020-04-15", - "announcementNumber": "DOE-20-ENG", - "hiringSubelementName": "Energy Research", - "positionTitle": "Backend Engineer", - "minimumGrade": "11", - "maximumGrade": "12", - "promotionPotential": None, - "minimumSalary": 80000.0, - "maximumSalary": 110000.0, - "supervisoryStatus": "Y", - "drugTestRequired": "Y", - "relocationExpensesReimbursed": "N", - "totalOpenings": "1", - "disableApplyOnline": "Y", - "positionOpeningStatus": "Closed", - "hiringPaths": [{"hiringPath": "Government employees"}], - "jobCategories": [{"series": "2210"}], - "positionLocations": [ - { - "positionLocationCity": "Washington", - "positionLocationState": "District of Columbia", - "positionLocationCountry": "United States", - } - ], - }, - ] + +@pytest.fixture +def make_historicjoa_response_payload( + make_historicjoa_item: Callable[..., dict[str, Any]], +) -> Callable[..., dict[str, Any]]: + """Return a factory that produces historic JOA endpoint responses.""" + + def _make( + *, + items: Sequence[dict[str, Any]] | None = None, + continuation_token: str | None = "NEXTTOKEN", + total_count: int = 2, + page_size: int = 2, + next_url: str | None = "https://example.invalid/historicjoa?page=2", + ) -> dict[str, Any]: + payload_items = ( + [deepcopy(item) for item in items] + if items is not None + else [ + make_historicjoa_item(base=0), + make_historicjoa_item(base=1), + ] + ) + metadata = { + "totalCount": total_count, + "pageSize": page_size, + "continuationToken": continuation_token, + } + return { + "paging": {"metadata": metadata, "next": next_url}, + "data": payload_items, + } + + return _make + + +@pytest.fixture +def historicjoa_response_payload( + make_historicjoa_response_payload: Callable[..., dict[str, Any]], +) -> dict[str, Any]: + """Serialized historic JOA response payload mimicking the USAJOBS API.""" + + return make_historicjoa_response_payload() diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index cd380f6..07863be 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,7 +1,5 @@ """Unit tests for USAJobsClient.""" -from copy import deepcopy - import pytest from usajobsapi.client import USAJobsClient @@ -12,47 +10,44 @@ # --- -def _build_search_payload(items, count, total=None): - """Build a serialized SearchResult payload for mocked responses.""" - - payload = { - "SearchResult": { - "SearchResultCount": count, - "SearchResultItems": items, - } - } - if total is not None: - payload["SearchResult"]["SearchResultCountAll"] = total - return payload - - -def test_search_jobs_pages_yields_pages(monkeypatch, search_result_item) -> None: +def test_search_jobs_pages_yields_pages( + monkeypatch: pytest.MonkeyPatch, + make_search_response_payload, + make_search_result_item, +) -> None: """Ensure search_jobs_pages iterates pages based on total counts.""" - first_item = deepcopy(search_result_item) - first_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 1 - second_item = deepcopy(search_result_item) - second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 2 - third_item = deepcopy(search_result_item) - third_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 3 - fourth_item = deepcopy(search_result_item) - fourth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 4 - fifth_item = deepcopy(search_result_item) - fifth_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 5 - responses = [ SearchEndpoint.Response.model_validate( - _build_search_payload([first_item, second_item], 2, total=5) + make_search_response_payload( + items=[ + make_search_result_item(matched_object_id=1), + make_search_result_item(matched_object_id=2), + ], + count=2, + total=5, + ) ), SearchEndpoint.Response.model_validate( - _build_search_payload([third_item, fourth_item], 2, total=5) + make_search_response_payload( + items=[ + make_search_result_item(matched_object_id=3), + make_search_result_item(matched_object_id=4), + ], + count=2, + total=5, + ) ), SearchEndpoint.Response.model_validate( - _build_search_payload([fifth_item], 1, total=5) + make_search_response_payload( + items=[make_search_result_item(matched_object_id=5)], + count=1, + total=5, + ) ), ] - captured_kwargs = [] + captured_kwargs: list[dict[str, object]] = [] def fake_search(self, **call_kwargs): captured_kwargs.append(call_kwargs) @@ -69,24 +64,31 @@ def fake_search(self, **call_kwargs): def test_search_jobs_pages_handles_missing_total( - monkeypatch, search_result_item + monkeypatch: pytest.MonkeyPatch, + make_search_response_payload, + make_search_result_item, ) -> None: """Continue until a short page is returned when total counts are absent.""" first_page = SearchEndpoint.Response.model_validate( - _build_search_payload( - [deepcopy(search_result_item), deepcopy(search_result_item)], - 2, + make_search_response_payload( + items=[ + make_search_result_item(matched_object_id=1), + make_search_result_item(matched_object_id=2), + ], + count=2, + include_total=False, ) ) - second_item = deepcopy(search_result_item) - second_item["MatchedObjectDescriptor"]["MatchedObjectId"] = 3 - second_page = SearchEndpoint.Response.model_validate( - _build_search_payload([second_item], 1) + final_page = SearchEndpoint.Response.model_validate( + make_search_response_payload( + items=[make_search_result_item(matched_object_id=3)], + count=1, + ) ) - responses = [first_page, second_page] - captured_kwargs = [] + responses = [first_page, final_page] + captured_kwargs: list[dict[str, object]] = [] def fake_search(self, **call_kwargs): captured_kwargs.append(call_kwargs) @@ -103,11 +105,14 @@ def fake_search(self, **call_kwargs): assert captured_kwargs[1]["results_per_page"] == 2 -def test_search_jobs_pages_breaks_on_empty_results(monkeypatch) -> None: +def test_search_jobs_pages_breaks_on_empty_results( + monkeypatch: pytest.MonkeyPatch, + make_search_response_payload, +) -> None: """Stop pagination when a page returns no results.""" empty_page = SearchEndpoint.Response.model_validate( - {"SearchResult": {"SearchResultCount": 0, "SearchResultItems": []}} + make_search_response_payload(items=[], count=0, include_total=False) ) def fake_search(self, **_): @@ -125,61 +130,37 @@ def fake_search(self, **_): # --- -def _search_response_payload( - items: list[dict | SearchEndpoint.JOAItem], - count: int, - total: int, - page: int, - results_per_page: int, -) -> dict: - return { - "LanguageCode": "EN", - "SearchParameters": { - "Page": page, - "ResultsPerPage": results_per_page, - }, - "SearchResult": { - "SearchResultCount": count, - "SearchResultCountAll": total, - "SearchResultItems": items, - }, - } - - -def test_search_jobs_items_yields_jobs(monkeypatch, search_result_item) -> None: +def test_search_jobs_items_yields_jobs( + monkeypatch: pytest.MonkeyPatch, + make_search_response_payload, + make_search_result_item, +) -> None: """Ensure search_jobs_items yields summaries across pages.""" client = USAJobsClient() - first_payload = deepcopy(search_result_item) - first_payload["MatchedObjectId"] = 1 - second_payload = deepcopy(search_result_item) - second_payload["MatchedObjectId"] = 2 - third_payload = deepcopy(search_result_item) - third_payload["MatchedObjectId"] = 3 - - responses = [ - _search_response_payload( - [ - first_payload, - second_payload, + payloads = [ + make_search_response_payload( + items=[ + make_search_result_item(matched_object_id=1), + make_search_result_item(matched_object_id=2), ], - 2, - 3, - 1, - 2, + count=2, + total=3, + page=1, + results_per_page=2, ), - _search_response_payload( - [third_payload], - 1, - 3, - 2, - 2, + make_search_response_payload( + items=[make_search_result_item(matched_object_id=3)], + count=1, + total=3, + page=2, + results_per_page=2, ), ] def fake_search_jobs(self, **_): - return SearchEndpoint.Response.model_validate(responses.pop(0)) + return SearchEndpoint.Response.model_validate(payloads.pop(0)) monkeypatch.setattr(USAJobsClient, "search_jobs", fake_search_jobs) @@ -193,20 +174,24 @@ def fake_search_jobs(self, **_): def test_historic_joa_pages_yields_pages( - monkeypatch, historicjoa_response_payload + monkeypatch: pytest.MonkeyPatch, + make_historicjoa_response_payload, ) -> None: """Ensure historic_joa_pages yields pages while forwarding continuation tokens.""" - first_payload = deepcopy(historicjoa_response_payload) - second_payload = deepcopy(historicjoa_response_payload) - second_payload["paging"]["metadata"]["continuationToken"] = None - second_payload["data"] = [] - responses = [ - HistoricJoaEndpoint.Response.model_validate(first_payload), - HistoricJoaEndpoint.Response.model_validate(second_payload), + HistoricJoaEndpoint.Response.model_validate( + make_historicjoa_response_payload() + ), + HistoricJoaEndpoint.Response.model_validate( + make_historicjoa_response_payload( + continuation_token=None, + items=[], + next_url=None, + ) + ), ] - captured_kwargs = [] + captured_kwargs: list[dict[str, object]] = [] def fake_historic_joa(self, **call_kwargs): captured_kwargs.append(call_kwargs) @@ -232,20 +217,21 @@ def fake_historic_joa(self, **call_kwargs): def test_historic_joa_pages_duplicate_token( - monkeypatch, historicjoa_response_payload + monkeypatch: pytest.MonkeyPatch, + make_historicjoa_response_payload, ) -> None: """Duplicate continuation tokens should raise to avoid infinite loops.""" first_response = HistoricJoaEndpoint.Response.model_validate( - historicjoa_response_payload - ) - duplicate_payload = deepcopy(historicjoa_response_payload) - duplicate_payload["paging"]["metadata"]["continuationToken"] = ( - first_response.next_token() + make_historicjoa_response_payload() ) responses = [ first_response, - HistoricJoaEndpoint.Response.model_validate(duplicate_payload), + HistoricJoaEndpoint.Response.model_validate( + make_historicjoa_response_payload( + continuation_token=first_response.next_token() + ) + ), ] def fake_historic_joa(self, **_): @@ -266,75 +252,35 @@ def fake_historic_joa(self, **_): def test_historic_joa_items_yields_items_across_pages( - monkeypatch: pytest.MonkeyPatch, historicjoa_response_payload + monkeypatch: pytest.MonkeyPatch, + make_historicjoa_response_payload, + make_historicjoa_item, ) -> None: """Ensure historic_joa_items yields items and follows continuation tokens.""" client = USAJobsClient() - first_page = deepcopy(historicjoa_response_payload) - first_page["paging"]["metadata"]["continuationToken"] = "TOKEN2" - first_page["data"] = first_page["data"][:2] - - second_page = { - "paging": { - "metadata": {"totalCount": 3, "pageSize": 1, "continuationToken": None}, - "next": None, - }, - "data": [ - { - "usajobsControlNumber": 111222333, - "hiringAgencyCode": "GSA", - "hiringAgencyName": "General Services Administration", - "hiringDepartmentCode": "GSA", - "hiringDepartmentName": "General Services Administration", - "agencyLevel": 1, - "agencyLevelSort": "GSA", - "appointmentType": "Permanent", - "workSchedule": "Full-time", - "payScale": "GS", - "salaryType": "Per Year", - "vendor": "USASTAFFING", - "travelRequirement": "Not required", - "teleworkEligible": "Y", - "serviceType": "Competitive", - "securityClearanceRequired": "N", - "securityClearance": "Not Required", - "whoMayApply": "All", - "announcementClosingTypeCode": "C", - "announcementClosingTypeDescription": "Closing Date", - "positionOpenDate": "2020-05-01", - "positionCloseDate": "2020-05-15", - "positionExpireDate": None, - "announcementNumber": "GSA-20-001", - "hiringSubelementName": "Administration", - "positionTitle": "Systems Analyst", - "minimumGrade": "11", - "maximumGrade": "12", - "promotionPotential": "13", - "minimumSalary": 85000.0, - "maximumSalary": 95000.0, - "supervisoryStatus": "N", - "drugTestRequired": "N", - "relocationExpensesReimbursed": "N", - "totalOpenings": "2", - "disableApplyOnline": "N", - "positionOpeningStatus": "Accepting Applications", - "hiringPaths": [{"hiringPath": "The public"}], - "jobCategories": [{"series": "2210"}], - "positionLocations": [ - { - "positionLocationCity": "Washington", - "positionLocationState": "District of Columbia", - "positionLocationCountry": "United States", - } - ], - } + first_page = make_historicjoa_response_payload(continuation_token="TOKEN2") + second_page = make_historicjoa_response_payload( + items=[ + make_historicjoa_item( + base=0, + usajobsControlNumber=111222333, + hiringAgencyCode="GSA", + hiringAgencyName="General Services Administration", + hiringDepartmentCode="GSA", + hiringDepartmentName="General Services Administration", + positionTitle="Systems Analyst", + ) ], - } + continuation_token=None, + total_count=3, + page_size=1, + next_url=None, + ) responses = [first_page, second_page] - calls = [] + calls: list[dict[str, object]] = [] def fake_historic(**call_kwargs): calls.append(call_kwargs) @@ -355,16 +301,15 @@ def fake_historic(**call_kwargs): def test_historic_joa_items_respects_initial_token( - monkeypatch: pytest.MonkeyPatch, historicjoa_response_payload + monkeypatch: pytest.MonkeyPatch, + make_historicjoa_response_payload, ) -> None: """Ensure historic_joa_items uses the supplied initial continuation token.""" client = USAJobsClient() - payload = deepcopy(historicjoa_response_payload) - payload["paging"]["metadata"]["continuationToken"] = None - - calls = [] + payload = make_historicjoa_response_payload(continuation_token=None) + calls: list[dict[str, object]] = [] def fake_historic(**call_kwargs): calls.append(call_kwargs)