diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..56d266a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,139 @@ +from typing import Dict, List + +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 historicjoa_response_payload() -> Dict[str, object]: + """Serialized Historic JOA response payload mimicking the USAJOBS API.""" + + return { + "paging": { + "metadata": { + "totalCount": 2, + "pageSize": 2, + "continuationToken": "NEXTTOKEN", + }, + "next": "https://example.invalid/historicjoa?page=2", + }, + "data": _historicjoa_items(), + } + + +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", + } + ], + }, + ] diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 877c54c..88565e7 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -1,6 +1,197 @@ -"""Placeholder tests for USAJobs API package.""" +"""Unit tests for USAJobsApiClient.""" +from copy import deepcopy -def test_placeholder() -> None: - """A trivial test to ensure the test suite runs.""" - assert True +import pytest + +from usajobsapi.client import USAJobsApiClient +from usajobsapi.endpoints.historicjoa import HistoricJoaEndpoint + +# test historic_joa_pages +# --- + + +def test_historic_joa_pages_yields_pages( + monkeypatch, 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), + ] + captured_kwargs = [] + + def fake_historic_joa(self, **call_kwargs): + captured_kwargs.append(call_kwargs) + return responses.pop(0) + + monkeypatch.setattr(USAJobsApiClient, "historic_joa", fake_historic_joa) + + client = USAJobsApiClient() + + pages = list( + client.historic_joa_pages( + hiring_agency_codes="NASA", continuation_token="INITIALTOKEN" + ) + ) + + assert len(pages) == 2 + assert pages[0].next_token() == "NEXTTOKEN" + assert pages[1].next_token() is None + assert captured_kwargs == [ + {"hiring_agency_codes": "NASA", "continuation_token": "INITIALTOKEN"}, + {"hiring_agency_codes": "NASA", "continuation_token": "NEXTTOKEN"}, + ] + + +def test_historic_joa_pages_duplicate_token( + monkeypatch, 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() + ) + responses = [ + first_response, + HistoricJoaEndpoint.Response.model_validate(duplicate_payload), + ] + + def fake_historic_joa(self, **_): + return responses.pop(0) + + monkeypatch.setattr(USAJobsApiClient, "historic_joa", fake_historic_joa) + + client = USAJobsApiClient() + iterator = client.historic_joa_pages() + + assert next(iterator) + with pytest.raises(RuntimeError, match="duplicate continuation token"): + next(iterator) + + +# test historic_joa_items +# --- + + +def test_historic_joa_items_yields_items_across_pages( + monkeypatch: pytest.MonkeyPatch, historicjoa_response_payload +) -> 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", + } + ], + } + ], + } + + responses = [first_page, second_page] + calls = [] + + def fake_historic(**call_kwargs): + calls.append(call_kwargs) + return HistoricJoaEndpoint.Response.model_validate(responses.pop(0)) + + monkeypatch.setattr(client, "historic_joa", fake_historic) + + items = list(client.historic_joa_items(hiring_agency_codes="NASA")) + + assert [item.usajobs_control_number for item in items] == [ + 123456789, + 987654321, + 111222333, + ] + assert len(calls) == 2 + assert "continuation_token" not in calls[0] + assert calls[1]["continuation_token"] == "TOKEN2" + + +def test_historic_joa_items_respects_initial_token( + monkeypatch: pytest.MonkeyPatch, 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 + + calls = [] + + def fake_historic(**call_kwargs): + calls.append(call_kwargs) + return HistoricJoaEndpoint.Response.model_validate(payload) + + monkeypatch.setattr(client, "historic_joa", fake_historic) + + items = list(client.historic_joa_items(continuation_token="SEED")) + + assert len(items) == len(payload["data"]) + assert calls[0]["continuation_token"] == "SEED" diff --git a/tests/unit/test_historicjoa.py b/tests/unit/test_historicjoa.py new file mode 100644 index 0000000..65d0b61 --- /dev/null +++ b/tests/unit/test_historicjoa.py @@ -0,0 +1,152 @@ +"""Unit tests for HistoricJoaEndpoint.""" + +import datetime as dt + +import pytest + +from usajobsapi.endpoints.historicjoa import HistoricJoaEndpoint + + +def test_params_to_params_serializes_aliases(historicjoa_params_kwargs) -> None: + """Validate Params.to_params uses USAJOBS aliases and formatting.""" + + params = HistoricJoaEndpoint.Params(**historicjoa_params_kwargs) + + assert isinstance(params.start_position_open_date, dt.date) + assert isinstance(params.end_position_open_date, dt.date) + assert isinstance(params.start_position_close_date, dt.date) + assert isinstance(params.end_position_close_date, dt.date) + + serialized = params.to_params() + + expected = { + "HiringAgencyCodes": "AGENCY1", + "HiringDepartmentCodes": "DEPT1", + "PositionSeries": "2210", + "AnnouncementNumbers": "23-ABC", + "USAJOBSControlNumbers": "1234567", + "StartPositionOpenDate": "2020-01-01", + "EndPositionOpenDate": "2020-12-31", + "StartPositionCloseDate": "2021-01-01", + "EndPositionCloseDate": "2021-12-31", + "continuationToken": "token123", + } + + assert serialized == expected + + +def test_params_to_params_accepts_datetime_for_start_open_date( + historicjoa_params_kwargs, +) -> None: + """Ensure start_position_open_date accepts datetime instances.""" + + kwargs = historicjoa_params_kwargs.copy() + kwargs["start_position_open_date"] = dt.datetime(2020, 1, 1, 8, 30) + + params = HistoricJoaEndpoint.Params(**kwargs) + + assert isinstance(params.start_position_open_date, dt.date) + + serialized = params.to_params() + + assert serialized["StartPositionOpenDate"] == "2020-01-01" + + +def test_params_to_params_rejects_time_for_start_open_date( + historicjoa_params_kwargs, +) -> None: + """Ensure start_position_open_date rejects time-only values.""" + + kwargs = historicjoa_params_kwargs.copy() + kwargs["start_position_open_date"] = dt.time(8, 45, 15) + + with pytest.raises(TypeError): + HistoricJoaEndpoint.Params(**kwargs) + + +def test_params_to_params_rejects_bad_string_format(historicjoa_params_kwargs) -> None: + """Ensure invalid date strings are rejected with a ValueError.""" + + kwargs = historicjoa_params_kwargs.copy() + kwargs["start_position_open_date"] = "01-01-2020" + + with pytest.raises(ValueError): + HistoricJoaEndpoint.Params(**kwargs) + + +def test_params_to_params_omits_none_fields(historicjoa_params_kwargs) -> None: + """Ensure Params.to_params excludes unset or None-valued fields.""" + + kwargs = historicjoa_params_kwargs.copy() + for optional in ( + "hiring_department_codes", + "announcement_numbers", + "continuation_token", + ): + kwargs[optional] = None + + params = HistoricJoaEndpoint.Params(**kwargs) + + assert params.start_position_open_date is None or isinstance( + params.start_position_open_date, dt.date + ) + + serialized = params.to_params() + + assert "HiringDepartmentCodes" not in serialized + assert "AnnouncementNumbers" not in serialized + assert "continuationToken" not in serialized + assert serialized["HiringAgencyCodes"] == "AGENCY1" + + +def test_item_model_parses_response_payload(historicjoa_response_payload) -> None: + """Confirm Item model accepts serialized payload dictionaries.""" + + payload = historicjoa_response_payload["data"][0] + + item = HistoricJoaEndpoint.Item.model_validate(payload) + + assert item.usajobs_control_number == 123456789 + assert item.hiring_agency_code == "NASA" + assert item.hiring_agency_name == "National Aeronautics and Space Administration" + assert item.hiring_department_code == "NAT" + assert item.hiring_department_name == "Department of Science" + assert item.agency_level == 2 + assert item.appointment_type == "Permanent" + assert item.position_title == "Data Scientist" + assert item.position_open_date == dt.date(2020, 1, 1) + assert item.position_close_date == dt.date(2020, 2, 1) + assert item.minimum_salary == 90000.0 + assert item.maximum_salary == 120000.0 + assert item.telework_eligible is True + assert item.security_clearance_required is True + assert item.security_clearance == "Secret" + assert item.supervisory_status is False + assert item.drug_test_required is False + assert item.relocation_expenses_reimbursed is True + assert item.disable_apply_online is False + assert len(item.hiring_paths) == 1 + assert item.hiring_paths[0].hiring_path == "The public" + assert len(item.job_categories) == 1 + assert item.job_categories[0].series == "1550" + assert len(item.position_locations) == 1 + location = item.position_locations[0] + assert location.position_location_city == "Houston" + assert location.position_location_state == "Texas" + assert location.position_location_country == "United States" + + +def test_response_next_token_returns_continuation(historicjoa_response_payload) -> None: + """Check Response.next_token surfaces continuation tokens from paging metadata.""" + + response = HistoricJoaEndpoint.Response.model_validate(historicjoa_response_payload) + + assert response.next_token() == "NEXTTOKEN" + + +def test_response_next_token_when_paging_missing() -> None: + """Validate Response.next_token returns None when paging metadata is absent.""" + + response = HistoricJoaEndpoint.Response(data=[]) + + assert response.next_token() is None diff --git a/tests/unit/test_utils.py b/tests/unit/test_utils.py index b03b5bf..fdb924e 100644 --- a/tests/unit/test_utils.py +++ b/tests/unit/test_utils.py @@ -1,10 +1,16 @@ +import datetime as dt from enum import StrEnum from typing import Annotated, Any, List, Optional import pytest from pydantic import BaseModel, Field -from usajobsapi.utils import _dump_by_alias, _normalize_param +from usajobsapi.utils import ( + _dump_by_alias, + _normalize_date, + _normalize_param, + _normalize_yn_bool, +) # testing models # --- @@ -33,6 +39,68 @@ class QueryModel(BaseModel): none_field: Annotated[Optional[str], Field(serialization_alias="G")] = None +# test _normalize_date +# --- + + +def test_normalize_date_accepts_datetime(): + dt_value = dt.datetime(2024, 5, 17, 15, 30, 45) + assert _normalize_date(dt_value) == dt.date(2024, 5, 17) + + +def test_normalize_date_accepts_date(): + date_value = dt.date(2024, 5, 17) + assert _normalize_date(date_value) == dt.date(2024, 5, 17) + + +def test_normalize_date_accepts_iso_string(): + assert _normalize_date("2024-05-17") == dt.date(2024, 5, 17) + + +def test_normalize_date_returns_none_for_none(): + assert _normalize_date(None) is None + + +def test_normalize_date_rejects_bad_string(): + with pytest.raises(ValueError): + _normalize_date("05/17/2024") + + +def test_normalize_date_rejects_non_date_inputs(): + with pytest.raises(TypeError): + _normalize_date(123) # pyright: ignore[reportArgumentType] + + +# test _normalize_yn_bool + + +def test_normalize_bool_accepts_bool(): + assert _normalize_yn_bool(True) + + +def test_normalize_bool_accepts_string(): + assert _normalize_yn_bool("Y") + assert _normalize_yn_bool("YES") + assert _normalize_yn_bool("TRUE") + assert not _normalize_yn_bool("N") + assert not _normalize_yn_bool("NO") + assert not _normalize_yn_bool("FALSE") + + +def test_normalize_bool_returns_none_for_none(): + assert _normalize_yn_bool(None) is None + + +def test_normalize_bool_rejects_bad_string(): + with pytest.raises(ValueError): + _normalize_yn_bool("indubitably") + + +def test_normalize_bool_rejects_non_bool_inputs(): + with pytest.raises(TypeError): + _normalize_yn_bool(123) # pyright: ignore[reportArgumentType] + + # test _normalize_param # --- diff --git a/usajobsapi/client.py b/usajobsapi/client.py index a2ed649..ce96f8c 100644 --- a/usajobsapi/client.py +++ b/usajobsapi/client.py @@ -1,5 +1,6 @@ """Wrapper for the USAJOBS REST API.""" +from collections.abc import Iterator from typing import Dict, Optional from urllib.parse import urlparse @@ -132,8 +133,86 @@ def historic_joa(self, **kwargs) -> HistoricJoaEndpoint.Response: """ params = HistoricJoaEndpoint.Params(**kwargs) resp = self._request( - HistoricJoaEndpoint.model_fields["method"].default, - HistoricJoaEndpoint.model_fields["path"].default, + HistoricJoaEndpoint.model_fields["METHOD"].default, + HistoricJoaEndpoint.model_fields["PATH"].default, params.to_params(), ) return HistoricJoaEndpoint.Response.model_validate(resp.json()) + + def historic_joa_pages(self, **kwargs) -> Iterator[HistoricJoaEndpoint.Response]: + """Yield Historic JOA pages, following continuation tokens. + + This can handle fresh requests or continue from a response page with a continuation token. + + :raises RuntimeError: On a duplicate continuation token. + :yield: The response object for the given continuation token. + :rtype: Iterator[HistoricJoaEndpoint.Response] + """ + + # Get the token by object name or alias name + token = kwargs.pop("continuationToken", kwargs.pop("continuation_token", None)) + + seen_tokens: set[str] = set() + if token: + seen_tokens.add(token) + + while True: + call_kwargs = kwargs + if token: + call_kwargs["continuation_token"] = token + + resp = self.historic_joa(**call_kwargs) + + next_token = resp.next_token() + # Handle duplicate tokens + if next_token and next_token in seen_tokens: + raise RuntimeError( + f"Historic JOA pagination returned duplicate continuation token '{next_token}'" + ) + + yield resp + + # If more pages + if not next_token: + break + + seen_tokens.add(next_token) + token = next_token + + def historic_joa_items(self, **kwargs) -> Iterator[HistoricJoaEndpoint.Item]: + """Yield Historic JOA items, following continuation tokens. + + This can handle fresh requests or continue from a response page with a continuation token. + + :raises RuntimeError: On a duplicate continuation token. + :yield: The response item. + :rtype: Iterator[HistoricJoaEndpoint.Item] + """ + token = kwargs.pop("continuationToken", kwargs.pop("continuation_token", None)) + + seen_tokens: set[str] = set() + if token: + seen_tokens.add(token) + + while True: + call_kwargs = kwargs + if token: + call_kwargs["continuation_token"] = token + + resp = self.historic_joa(**call_kwargs) + + for item in resp.data: + yield item + + next_token = resp.next_token() + # Handle duplicate tokens + if next_token and next_token in seen_tokens: + raise RuntimeError( + f"Historic JOA pagination returned duplicate continuation token '{next_token}'" + ) + + if not next_token: + break + + seen_tokens.add(next_token) + token = next_token diff --git a/usajobsapi/endpoints/historicjoa.py b/usajobsapi/endpoints/historicjoa.py index 4f7f04e..60ac8e3 100644 --- a/usajobsapi/endpoints/historicjoa.py +++ b/usajobsapi/endpoints/historicjoa.py @@ -1,19 +1,229 @@ """Wrapper for the Historic JOAs API.""" -from typing import Dict +import datetime as dt +from typing import Dict, List, Optional -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict, Field, field_validator -from usajobsapi.utils import _dump_by_alias +from usajobsapi.utils import _dump_by_alias, _normalize_date, _normalize_yn_bool class HistoricJoaEndpoint(BaseModel): - method: str = "GET" - path: str = "/api/historicjoa" + """ + Declarative wrapper around the [Historic JOAs API](https://developer.usajobs.gov/api-reference/get-api-historicjoa). + """ + + METHOD: str = "GET" + PATH: str = "/api/historicjoa" class Params(BaseModel): + """Declarative definition of the endpoint's query parameters.""" + + model_config = ConfigDict(frozen=True, extra="forbid", populate_by_name=True) + + hiring_agency_codes: Optional[str] = Field( + None, serialization_alias="HiringAgencyCodes" + ) + hiring_department_codes: Optional[str] = Field( + None, serialization_alias="HiringDepartmentCodes" + ) + position_series: Optional[str] = Field( + None, serialization_alias="PositionSeries" + ) + announcement_numbers: Optional[str] = Field( + None, serialization_alias="AnnouncementNumbers" + ) + usajobs_control_numbers: Optional[str] = Field( + None, serialization_alias="USAJOBSControlNumbers" + ) + start_position_open_date: Optional[dt.date] = Field( + default=None, serialization_alias="StartPositionOpenDate" + ) + end_position_open_date: Optional[dt.date] = Field( + default=None, serialization_alias="EndPositionOpenDate" + ) + start_position_close_date: Optional[dt.date] = Field( + default=None, serialization_alias="StartPositionCloseDate" + ) + end_position_close_date: Optional[dt.date] = Field( + default=None, serialization_alias="EndPositionCloseDate" + ) + continuation_token: Optional[str] = Field( + None, serialization_alias="continuationToken" + ) + def to_params(self) -> Dict[str, str]: + """Serialize params into payload-ready query parameters.""" return _dump_by_alias(self) + @field_validator( + "start_position_open_date", + "end_position_open_date", + "start_position_close_date", + "end_position_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) + + # Response shapes + # --- + + class Item(BaseModel): + """A single historic job opportunity announcement record.""" + + class HiringPath(BaseModel): + hiring_path: Optional[str] = Field(default=None, alias="hiringPath") + + class JobCategory(BaseModel): + series: Optional[str] = Field(default=None, alias="series") + + class PositionLocation(BaseModel): + position_location_city: Optional[str] = Field( + default=None, alias="positionLocationCity" + ) + position_location_state: Optional[str] = Field( + default=None, alias="positionLocationState" + ) + position_location_country: Optional[str] = Field( + default=None, alias="positionLocationCountry" + ) + + usajobs_control_number: int = Field(alias="usajobsControlNumber") + hiring_agency_code: Optional[str] = Field( + default=None, alias="hiringAgencyCode" + ) + hiring_agency_name: Optional[str] = Field( + default=None, alias="hiringAgencyName" + ) + hiring_department_code: Optional[str] = Field( + default=None, alias="hiringDepartmentCode" + ) + hiring_department_name: Optional[str] = Field( + default=None, alias="hiringDepartmentName" + ) + agency_level: Optional[int] = Field(default=None, alias="agencyLevel") + agency_level_sort: Optional[str] = Field(default=None, alias="agencyLevelSort") + appointment_type: Optional[str] = Field(default=None, alias="appointmentType") + work_schedule: Optional[str] = Field(default=None, alias="workSchedule") + pay_scale: Optional[str] = Field(default=None, alias="payScale") + salary_type: Optional[str] = Field(default=None, alias="salaryType") + vendor: Optional[str] = Field(default=None, alias="vendor") + travel_requirement: Optional[str] = Field( + default=None, alias="travelRequirement" + ) + telework_eligible: Optional[bool] = Field( + default=None, alias="teleworkEligible" + ) + service_type: Optional[str] = Field(default=None, alias="serviceType") + security_clearance_required: Optional[bool] = Field( + default=None, alias="securityClearanceRequired" + ) + security_clearance: Optional[str] = Field( + default=None, alias="securityClearance" + ) + who_may_apply: Optional[str] = Field(default=None, alias="whoMayApply") + announcement_closing_type_code: Optional[str] = Field( + default=None, alias="announcementClosingTypeCode" + ) + announcement_closing_type_description: Optional[str] = Field( + default=None, alias="announcementClosingTypeDescription" + ) + position_open_date: Optional[dt.date] = Field( + default=None, alias="positionOpenDate" + ) + position_close_date: Optional[dt.date] = Field( + default=None, alias="positionCloseDate" + ) + position_expire_date: Optional[dt.date] = Field( + default=None, alias="positionExpireDate" + ) + announcement_number: Optional[str] = Field( + default=None, alias="announcementNumber" + ) + hiring_subelement_name: Optional[str] = Field( + default=None, alias="hiringSubelementName" + ) + position_title: Optional[str] = Field(default=None, alias="positionTitle") + minimum_grade: Optional[str] = Field(default=None, alias="minimumGrade") + maximum_grade: Optional[str] = Field(default=None, alias="maximumGrade") + promotion_potential: Optional[str] = Field( + default=None, alias="promotionPotential" + ) + minimum_salary: Optional[float] = Field(default=None, alias="minimumSalary") + maximum_salary: Optional[float] = Field(default=None, alias="maximumSalary") + supervisory_status: Optional[bool] = Field( + default=None, alias="supervisoryStatus" + ) + drug_test_required: Optional[bool] = Field( + default=None, alias="drugTestRequired" + ) + relocation_expenses_reimbursed: Optional[bool] = Field( + default=None, alias="relocationExpensesReimbursed" + ) + total_openings: Optional[str] = Field(default=None, alias="totalOpenings") + disable_apply_online: Optional[bool] = Field( + default=None, alias="disableApplyOnline" + ) + position_opening_status: Optional[str] = Field( + default=None, alias="positionOpeningStatus" + ) + hiring_paths: List["HistoricJoaEndpoint.Item.HiringPath"] = Field( + default_factory=list, alias="hiringPaths" + ) + job_categories: List["HistoricJoaEndpoint.Item.JobCategory"] = Field( + default_factory=list, alias="jobCategories" + ) + position_locations: List["HistoricJoaEndpoint.Item.PositionLocation"] = Field( + default_factory=list, alias="positionLocations" + ) + + @field_validator( + "telework_eligible", + "security_clearance_required", + "supervisory_status", + "drug_test_required", + "relocation_expenses_reimbursed", + "disable_apply_online", + mode="before", + ) + @classmethod + def _normalize_yn_boolean(cls, value: None | bool | str) -> Optional[bool]: + """Coerce bool-like outputs to `bool`.""" + + return _normalize_yn_bool(value) + + class PagingMeta(BaseModel): + """Pagination metadata returned alongside Historic JOA results.""" + + total_count: Optional[int] = Field(default=None, alias="totalCount") + page_size: Optional[int] = Field(default=None, alias="pageSize") + continuation_token: Optional[str] = Field( + default=None, alias="continuationToken" + ) + + class Paging(BaseModel): + """Container for pagination metadata and optional navigation links.""" + + metadata: "HistoricJoaEndpoint.PagingMeta" + next: Optional[str] = None + class Response(BaseModel): - pass + """Declarative definition of the endpoint's response object.""" + + paging: Optional["HistoricJoaEndpoint.Paging"] = None + data: List["HistoricJoaEndpoint.Item"] = Field(default_factory=list) + + def next_token(self) -> Optional[str]: + """Return the continuation token for requesting the next page.""" + + return ( + self.paging.metadata.continuation_token + if self.paging and self.paging.metadata + else None + ) diff --git a/usajobsapi/utils.py b/usajobsapi/utils.py index 3e114a9..ef63e40 100644 --- a/usajobsapi/utils.py +++ b/usajobsapi/utils.py @@ -1,11 +1,48 @@ """Helper utility functions.""" +import datetime as dt from enum import Enum from typing import Any, Dict, Optional from pydantic import BaseModel +def _normalize_date(value: None | dt.datetime | dt.date | str) -> Optional[dt.date]: + """Normalize to `datetime.date`.""" + + if value is None: + return None + if isinstance(value, dt.datetime): + return value.date() + if isinstance(value, dt.date): + return value + if isinstance(value, str): + try: + return dt.date.fromisoformat(value) + except ValueError as exc: + msg = "Value must be an ISO 8601 date string (YYYY-MM-DD)" + raise ValueError(msg) from exc + msg = "Expected value type of datetime, date, or ISO date string" + raise TypeError(msg) + + +def _normalize_yn_bool(value: None | bool | str) -> Optional[bool]: + """Normalize "Y"/"N" to `bool`.""" + + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().upper() + if normalized in {"Y", "YES", "TRUE", "1"}: + return True + if normalized in {"N", "NO", "FALSE", "0"}: + return False + raise ValueError("Value must be a Y/YES/TRUE/1 or N/NO/FALSE/0 string") + raise TypeError("Expected value type of bool or Y/N string") + + def _normalize_param(value: Any) -> Optional[str]: """Normalize query parameters to the format expected by USAJOBS.