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 diff --git a/tests/conftest.py b/tests/conftest.py index 56d266a..049accb 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,11 +1,311 @@ -from typing import Dict, List +from collections.abc import Callable, Sequence +from copy import deepcopy +from typing import Any import pytest +from usajobsapi.endpoints.search import HiringPath + +_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 historicjoa_params_kwargs() -> Dict[str, str]: - """Field-value mapping used to build HistoricJoaEndpoint 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"], + "radius": 25, + "relocation": True, + "job_category_codes": ["001", "002"], + "hiring_paths": [HiringPath.PUBLIC, HiringPath.VET], + } + + +@pytest.fixture +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_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 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 + + +@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.""" return { "hiring_agency_codes": "AGENCY1", @@ -22,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 88565e7..07863be 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,39 +1,205 @@ -"""Unit tests for USAJobsApiClient.""" - -from copy import deepcopy +"""Unit tests for USAJobsClient.""" import pytest -from usajobsapi.client import USAJobsApiClient +from usajobsapi.client import USAJobsClient from usajobsapi.endpoints.historicjoa import HistoricJoaEndpoint +from usajobsapi.endpoints.search import SearchEndpoint + +# test search_jobs_pages +# --- + + +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.""" + + responses = [ + SearchEndpoint.Response.model_validate( + 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( + 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( + make_search_response_payload( + items=[make_search_result_item(matched_object_id=5)], + count=1, + total=5, + ) + ), + ] + + captured_kwargs: list[dict[str, object]] = [] + + 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: 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( + 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, + ) + ) + final_page = SearchEndpoint.Response.model_validate( + make_search_response_payload( + items=[make_search_result_item(matched_object_id=3)], + count=1, + ) + ) + + responses = [first_page, final_page] + captured_kwargs: list[dict[str, object]] = [] + + 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: pytest.MonkeyPatch, + make_search_response_payload, +) -> None: + """Stop pagination when a page returns no results.""" + + empty_page = SearchEndpoint.Response.model_validate( + make_search_response_payload(items=[], count=0, include_total=False) + ) + + 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 search_jobs_items +# --- + + +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() + + payloads = [ + make_search_response_payload( + items=[ + make_search_result_item(matched_object_id=1), + make_search_result_item(matched_object_id=2), + ], + count=2, + total=3, + page=1, + results_per_page=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(payloads.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 # --- 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) 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( @@ -51,28 +217,29 @@ 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, **_): 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) @@ -85,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 = USAJobsApiClient() - - 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", - } - ], - } + client = USAJobsClient() + + 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) @@ -174,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 = USAJobsApiClient() - - payload = deepcopy(historicjoa_response_payload) - payload["paging"]["metadata"]["continuationToken"] = None + client = USAJobsClient() - calls = [] + payload = make_historicjoa_response_payload(continuation_token=None) + calls: list[dict[str, object]] = [] def fake_historic(**call_kwargs): calls.append(call_kwargs) diff --git a/tests/unit/test_search.py b/tests/unit/test_search.py new file mode 100644 index 0000000..9aa2e36 --- /dev/null +++ b/tests/unit/test_search.py @@ -0,0 +1,97 @@ +import pytest + +from usajobsapi.endpoints.search import SearchEndpoint + + +class TestSearchEndpointParams: + 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", + "RelocationIndicator": "True", + "JobCategoryCode": "001;002", + "HiringPath": "public;vet", + } + + assert serialized == expected + + 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_job_summary_parses_nested_fields(self, search_result_item): + summary = SearchEndpoint.JOAItem.model_validate(search_result_item) + + 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.details.department_name + == "National Aeronautics and Space Administration" + ) + assert summary.details.locations_display == "Houston, TX" + + 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.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( + search_result_payload + ) + + 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_response_payload): + resp = SearchEndpoint.Response.model_validate(search_response_payload) + + 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 + + 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/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/__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 ce96f8c..7f87907 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__( @@ -110,21 +110,90 @@ 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) 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, ) return SearchEndpoint.Response.model_validate(resp.json()) + def search_jobs_pages(self, **kwargs) -> Iterator[SearchEndpoint.Response]: + """Yield Job Search 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 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.JOAItem] + """ + 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. 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 7159f3c..966cabb 100644 --- a/usajobsapi/endpoints/search.py +++ b/usajobsapi/endpoints/search.py @@ -1,19 +1,762 @@ """Wrapper for the Job Search API.""" -from typing import Dict +from __future__ import annotations -from pydantic import BaseModel +import datetime as dt +from enum import StrEnum +from typing import Annotated, Any, Dict, List, Optional -from usajobsapi.utils import _dump_by_alias +from pydantic import ( + AfterValidator, + BaseModel, + ConfigDict, + Field, + ValidationInfo, + field_validator, + model_validator, +) +from usajobsapi.endpoints._validators import isvalid_pay_grade +from usajobsapi.utils import _dump_by_alias, _normalize_date +# Enums for query-params +# --- + + +class SortField(StrEnum): + """Sort the search by the specified field.""" + + 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): + """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 FieldsMinMax(StrEnum): + """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" + 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): - method: str = "GET" - path: str = "/api/search" + """ + 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 of the endpoint's query parameters.""" + + model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) + + keyword: Optional[str] = Field( + None, + 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, + 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, + 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, + 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, + 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, + 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", + ], + ) + ) + + job_category_codes: List[str] = Field( + default_factory=list, + 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, + 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, + 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, + 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, + 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", + ], + ) + + travel_percentage: List[int] = Field( + default_factory=list, + 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, + alias="RelocationIndicator", + description="Issues a search to find hits for jobs matching the relocation filter.", + ) + security_clearance_required: List[int] = Field( + default_factory=list, + 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, + alias="SupervisoryStatus", + description="Issues a search to find hits for jobs matching the specified supervisory status.", + ) + + days_since_posted: Annotated[ + Optional[int], + Field( + 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, + 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, + 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, + alias="SortDirection", + description="Issues a search that will be sorted by the SortField specified, in the direction specified.", + ) + + page: Annotated[ + Optional[int], + Field( + alias="Page", + description="Issues a search to pull the paged results specified.", + strict=True, + ge=1, + ), + ] = None + results_per_page: Annotated[ + Optional[int], + Field( + 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, + 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( + 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[FieldsMinMax] = Field( + None, + 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, + 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, + 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, + 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, + 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, + 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, + alias="RemoteIndicator", + description="Issues a search to find hits for jobs matching the remote filter.", + ) + + @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 + + @field_validator("remuneration_max") + @classmethod + 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: + raise ValueError( + "RemunerationMaximumAmount must be >= RemunerationMinimumAmount." + ) + return v + def to_params(self) -> Dict[str, str]: + """Return the serialized query-parameter dictionary.""" return _dump_by_alias(self) + # 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 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.""" + + 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", + 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): + """Wrapper for additional USAJOBS metadata.""" + + 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 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.", + ) + + locations_display: Optional[str] = Field( + default=None, alias="PositionLocationDisplay" + ) + locations: List[SearchEndpoint.JobLocation] = Field( + 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", + description="List of job category objects that contain values for name and code.", + ) + job_grades: List[SearchEndpoint.JobGrade] = Field( + 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", + description="List of position schedule objects that contains values for name and code.", + ) + position_offerings: List[SearchEndpoint.PositionOfferingType] = Field( + 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", + description="Summary of qualifications for the job offering.", + ) + + position_remuneration: List[SearchEndpoint.PositionRemuneration] = Field( + 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", + description="Date the job opportunity is posted", + ) + application_close_date: Optional[dt.date] = Field( + 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( + "position_start_date", + "position_end_date", + "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.""" + + if self.user_area and self.user_area.details: + details = self.user_area.details + if self.user_area.details and details.job_summary: + return details.job_summary + return self.qualification_summary + + class JOAItem(BaseModel): + """Represents a job opportunity annoucement object search result item.""" + + 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.""" + + 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[SearchEndpoint.JOAItem] = Field( + default_factory=list, + alias="SearchResultItems", + description="Array of job opportunity announcement objects that matched search criteria.", + ) + class Response(BaseModel): - pass + """Declarative definition of the endpoint's response object.""" + + language: Optional[str] = Field( + default=None, alias="LanguageCode", description="Response Langauge" + ) + params: Optional[SearchEndpoint.Params] = Field( + 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" + ) + + def jobs(self) -> List[SearchEndpoint.JOAItem]: + """Helper method to directly expose search result items from the response object.""" + if not self.search_result: + return [] + return self.search_result.items 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