Skip to content

API reference

Auto-generated from docstrings.

Client

sjvair.client.SJVAirClient

HTTP client for the SJVAir API.

All resource objects (monitors, regions, calenviroscreen, ceidars, hms, pesticides) are attached as attributes and share this client's session, retry logic, and cooldown gate.

Parameters:

Name Type Description Default
base_url str | None

API base URL. Defaults to SJVAIR_BASE_URL env var or the production URL.

 None
timeout int | None

Request timeout in seconds. Defaults to SJVAIR_TIMEOUT env var or 30.

 None
max_retries int | None

Number of retries on 5xx / 429 responses. Defaults to 5.

 None
max_connections int | None

Maximum concurrent requests (semaphore). Defaults to 4.

 None
api_key str | None

Bearer token for authenticated endpoints. Defaults to SJVAIR_API_KEY env var.

 None

Can be used as a context manager to ensure the underlying session is closed::

with SJVAirClient() as client:
    monitors = list(client.monitors.list())
Source code in sjvair/client.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
class SJVAirClient:
    """HTTP client for the SJVAir API.

    All resource objects (``monitors``, ``regions``, ``calenviroscreen``, ``ceidars``,
    ``hms``, ``pesticides``) are attached as attributes and share this client's session,
    retry logic, and cooldown gate.

    Args:
        base_url: API base URL. Defaults to ``SJVAIR_BASE_URL`` env var or the production URL.
        timeout: Request timeout in seconds. Defaults to ``SJVAIR_TIMEOUT`` env var or 30.
        max_retries: Number of retries on 5xx / 429 responses. Defaults to 5.
        max_connections: Maximum concurrent requests (semaphore). Defaults to 4.
        api_key: Bearer token for authenticated endpoints. Defaults to ``SJVAIR_API_KEY`` env var.

    Can be used as a context manager to ensure the underlying session is closed::

        with SJVAirClient() as client:
            monitors = list(client.monitors.list())
    """

    RETRYABLE = frozenset({500, 502, 503, 504})

    def __init__(
        self,
        base_url: str | None = None,
        timeout: int | None = None,
        max_retries: int | None = None,
        max_connections: int | None = None,
        api_key: str | None = None,
    ) -> None:
        self.base_url = (base_url or os.environ.get('SJVAIR_BASE_URL') or DEFAULT_BASE_URL).rstrip('/') + '/'
        self.timeout = int(timeout if timeout is not None else os.environ.get('SJVAIR_TIMEOUT') or DEFAULT_TIMEOUT)
        self.max_retries = int(max_retries if max_retries is not None else DEFAULT_MAX_RETRIES)
        self.api_key = api_key or os.environ.get('SJVAIR_API_KEY')

        self._semaphore = threading.BoundedSemaphore(int(max_connections or DEFAULT_MAX_CONNECTIONS))
        self._cooldown = CooldownGate()
        self._session = self._build_session()

        from .resources.calenviroscreen import CalEnviroScreenResource
        from .resources.ceidars import CEIDARSResource
        from .resources.hms import HMSResource
        from .resources.monitors import MonitorsResource
        from .resources.pesticides import PesticidesResource
        from .resources.regions import RegionsResource

        self.monitors = MonitorsResource(self)
        self.regions = RegionsResource(self)
        self.calenviroscreen = CalEnviroScreenResource(self)
        self.ceidars = CEIDARSResource(self)
        self.hms = HMSResource(self)
        self.pesticides = PesticidesResource(self)

    def _build_session(self) -> requests.Session:
        session = requests.Session()
        if self.api_key:
            session.headers['Authorization'] = f'Bearer {self.api_key}'
        return session

    def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
        """GET ``path`` relative to ``base_url``, with retry and cooldown.

        Retries on 5xx up to ``max_retries`` times with exponential backoff.
        On 429, triggers a shared cooldown that blocks all threads until the
        wait expires. Raises :class:`~sjvair.exceptions.NotFound` on 404,
        :class:`~sjvair.exceptions.RateLimited` after exhausting retries on 429,
        and :class:`~sjvair.exceptions.ServerError` after exhausting retries on 5xx.
        """
        url = self.base_url + path.lstrip('/')
        self._cooldown.wait()
        with self._semaphore:
            last_exc: Exception = RuntimeError('no attempts made')
            for attempt in range(self.max_retries + 1):
                try:
                    log.debug('GET %s params=%s attempt=%d', url, params, attempt)
                    r = self._session.get(url, params=params, timeout=self.timeout)
                    if r.status_code == 404:
                        raise NotFound(f'Not found: {url}')
                    if r.status_code == 429:
                        raise RateLimited(
                            f'Rate limited: {url}',
                            retry_after=float(r.headers.get('Retry-After', 60)),
                        )
                    if r.status_code in self.RETRYABLE:
                        raise ServerError(f'HTTP {r.status_code}: {url}')
                    r.raise_for_status()
                    return r.json()
                except RateLimited as exc:
                    last_exc = exc
                    if attempt >= self.max_retries:
                        raise
                    delay = (exc.retry_after or 60) * (2**attempt)
                    log.warning('Rate limited; cooling down %.1fs', delay)
                    self._cooldown.cooldown(delay)
                except ServerError as exc:
                    last_exc = exc
                    if attempt >= self.max_retries:
                        raise
                    delay = float(2**attempt)
                    log.warning('Server error attempt %d; retry in %.1fs', attempt + 1, delay)
                    time.sleep(delay)
            raise last_exc

    def close(self) -> None:
        self._session.close()

    def __enter__(self) -> SJVAirClient:
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

get 

get(path, params=None)

GET path relative to base_url, with retry and cooldown.

Retries on 5xx up to max_retries times with exponential backoff. On 429, triggers a shared cooldown that blocks all threads until the wait expires. Raises :class:~sjvair.exceptions.NotFound on 404, :class:~sjvair.exceptions.RateLimited after exhausting retries on 429, and :class:~sjvair.exceptions.ServerError after exhausting retries on 5xx.

Source code in sjvair/client.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
    """GET ``path`` relative to ``base_url``, with retry and cooldown.

    Retries on 5xx up to ``max_retries`` times with exponential backoff.
    On 429, triggers a shared cooldown that blocks all threads until the
    wait expires. Raises :class:`~sjvair.exceptions.NotFound` on 404,
    :class:`~sjvair.exceptions.RateLimited` after exhausting retries on 429,
    and :class:`~sjvair.exceptions.ServerError` after exhausting retries on 5xx.
    """
    url = self.base_url + path.lstrip('/')
    self._cooldown.wait()
    with self._semaphore:
        last_exc: Exception = RuntimeError('no attempts made')
        for attempt in range(self.max_retries + 1):
            try:
                log.debug('GET %s params=%s attempt=%d', url, params, attempt)
                r = self._session.get(url, params=params, timeout=self.timeout)
                if r.status_code == 404:
                    raise NotFound(f'Not found: {url}')
                if r.status_code == 429:
                    raise RateLimited(
                        f'Rate limited: {url}',
                        retry_after=float(r.headers.get('Retry-After', 60)),
                    )
                if r.status_code in self.RETRYABLE:
                    raise ServerError(f'HTTP {r.status_code}: {url}')
                r.raise_for_status()
                return r.json()
            except RateLimited as exc:
                last_exc = exc
                if attempt >= self.max_retries:
                    raise
                delay = (exc.retry_after or 60) * (2**attempt)
                log.warning('Rate limited; cooling down %.1fs', delay)
                self._cooldown.cooldown(delay)
            except ServerError as exc:
                last_exc = exc
                if attempt >= self.max_retries:
                    raise
                delay = float(2**attempt)
                log.warning('Server error attempt %d; retry in %.1fs', attempt + 1, delay)
                time.sleep(delay)
        raise last_exc

Resources

sjvair.resources.monitors.MonitorsResource

Bases: BaseResource

Access air quality monitor data.

Available on :attr:SJVAirClient.monitors.

Source code in sjvair/resources/monitors.py 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
class MonitorsResource(BaseResource):
    """Access air quality monitor data.

    Available on :attr:`SJVAirClient.monitors`.
    """

    PATH = 'monitors/'

    def list(self, **params: Any) -> Iterator[dict[str, Any]]:
        """Iterate all monitors, optionally filtered by ``region_id``, ``is_sjvair``, etc."""
        return self._paginate(self.PATH, params or None)

    def get(self, monitor_id: str) -> dict[str, Any]:
        """Get a single monitor by ID."""
        return self._client.get(f'{self.PATH}{monitor_id}/')['data']

    def meta(self) -> dict[str, Any]:
        """Return field metadata for monitor entries (field names, units, etc.)."""
        return self._client.get(f'{self.PATH}meta/')['data']

    def entries(self, monitor_id: str, entry_type: str, **params: Any) -> Iterator[dict[str, Any]]:
        """Iterate paginated entries for one monitor and entry type (e.g. ``'PM2.5'``)."""
        return self._paginate(f'{self.PATH}{monitor_id}/entries/{entry_type}/', params or None)

    def export(
        self,
        monitor_id: str,
        start_date: str,
        end_date: str,
        scope: str = 'resolved',
    ) -> Iterator[dict[str, Any]]:
        """Bulk-export entries for a monitor in a single request.

        The server enforces a 180-day maximum window per call. Use
        :class:`~sjvair.export.engine.ExportEngine` to download longer ranges
        automatically by splitting into chunks.

        Args:
            monitor_id: Monitor UUID.
            start_date: ISO 8601 date string (``YYYY-MM-DD``).
            end_date: ISO 8601 date string (``YYYY-MM-DD``).
            scope: ``'resolved'`` (calibrated) or ``'expanded'`` (raw + derived fields).
        """
        data = self._client.get(
            f'{self.PATH}{monitor_id}/entries/export/json/',
            {'start_date': start_date, 'end_date': end_date, 'scope': scope},
        )
        return iter(data['data'])

    def summaries(
        self,
        monitor_id: str,
        entry_type: str,
        resolution: str,
        start_date: str,
        end_date: str,
    ) -> Iterator[dict[str, Any]]:
        """Iterate aggregated summaries for a monitor across the given date range.

        Args:
            monitor_id: Monitor UUID.
            entry_type: Sensor field (e.g. ``'PM2.5'``).
            resolution: One of ``'hourly'``, ``'daily'``, ``'monthly'``,
                ``'quarterly'``, ``'seasonal'``, ``'yearly'``.
            start_date: ISO 8601 date string.
            end_date: ISO 8601 date string.
        """
        base = f'{self.PATH}{monitor_id}/summaries/'
        paths = _iter_summary_paths(
            base,
            entry_type,
            resolution,
            date.fromisoformat(start_date),
            date.fromisoformat(end_date),
        )
        rows = itertools.chain.from_iterable(self._paginate(p) for p in paths)
        # The API's summary rows don't identify their monitor; tag each row so
        # callers fanning out across monitors can attribute results.
        return ({'monitor_id': monitor_id, **row} for row in rows)

    def closest(self, entry_type: str, lat: float, lon: float) -> list[dict[str, Any]]:  # ty: ignore[invalid-type-form]
        """Return up to 3 nearest active monitors with distance and latest entry."""
        return self._client.get(f'monitors/{entry_type}/closest/', {'lat': lat, 'lon': lon})['data']

    def current(self, entry_type: str) -> Iterator[dict[str, Any]]:
        """Iterate all active monitors with their most recent entry for the given type."""
        return self._paginate(f'monitors/{entry_type}/current/')

list 

list(**params)

Iterate all monitors, optionally filtered by region_id, is_sjvair, etc.

Source code in sjvair/resources/monitors.py 
44
45
46
def list(self, **params: Any) -> Iterator[dict[str, Any]]:
    """Iterate all monitors, optionally filtered by ``region_id``, ``is_sjvair``, etc."""
    return self._paginate(self.PATH, params or None)

get 

get(monitor_id)

Get a single monitor by ID.

Source code in sjvair/resources/monitors.py 
48
49
50
def get(self, monitor_id: str) -> dict[str, Any]:
    """Get a single monitor by ID."""
    return self._client.get(f'{self.PATH}{monitor_id}/')['data']

meta 

meta()

Return field metadata for monitor entries (field names, units, etc.).

Source code in sjvair/resources/monitors.py 
52
53
54
def meta(self) -> dict[str, Any]:
    """Return field metadata for monitor entries (field names, units, etc.)."""
    return self._client.get(f'{self.PATH}meta/')['data']

entries 

entries(monitor_id, entry_type, **params)

Iterate paginated entries for one monitor and entry type (e.g. 'PM2.5').

Source code in sjvair/resources/monitors.py 
56
57
58
def entries(self, monitor_id: str, entry_type: str, **params: Any) -> Iterator[dict[str, Any]]:
    """Iterate paginated entries for one monitor and entry type (e.g. ``'PM2.5'``)."""
    return self._paginate(f'{self.PATH}{monitor_id}/entries/{entry_type}/', params or None)

export 

export(monitor_id, start_date, end_date, scope='resolved')

Bulk-export entries for a monitor in a single request.

The server enforces a 180-day maximum window per call. Use :class:~sjvair.export.engine.ExportEngine to download longer ranges automatically by splitting into chunks.

Parameters:

Name Type Description Default
monitor_id str

Monitor UUID.

 required
start_date str

ISO 8601 date string (YYYY-MM-DD).

 required
end_date str

ISO 8601 date string (YYYY-MM-DD).

 required
scope str

'resolved' (calibrated) or 'expanded' (raw + derived fields).

 'resolved'
Source code in sjvair/resources/monitors.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def export(
    self,
    monitor_id: str,
    start_date: str,
    end_date: str,
    scope: str = 'resolved',
) -> Iterator[dict[str, Any]]:
    """Bulk-export entries for a monitor in a single request.

    The server enforces a 180-day maximum window per call. Use
    :class:`~sjvair.export.engine.ExportEngine` to download longer ranges
    automatically by splitting into chunks.

    Args:
        monitor_id: Monitor UUID.
        start_date: ISO 8601 date string (``YYYY-MM-DD``).
        end_date: ISO 8601 date string (``YYYY-MM-DD``).
        scope: ``'resolved'`` (calibrated) or ``'expanded'`` (raw + derived fields).
    """
    data = self._client.get(
        f'{self.PATH}{monitor_id}/entries/export/json/',
        {'start_date': start_date, 'end_date': end_date, 'scope': scope},
    )
    return iter(data['data'])

summaries 

summaries(
    monitor_id, entry_type, resolution, start_date, end_date
)

Iterate aggregated summaries for a monitor across the given date range.

Parameters:

Name Type Description Default
monitor_id str

Monitor UUID.

 required
entry_type str

Sensor field (e.g. 'PM2.5').

 required
resolution str

One of 'hourly', 'daily', 'monthly', 'quarterly', 'seasonal', 'yearly'.

 required
start_date str

ISO 8601 date string.

 required
end_date str

ISO 8601 date string.

 required
Source code in sjvair/resources/monitors.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def summaries(
    self,
    monitor_id: str,
    entry_type: str,
    resolution: str,
    start_date: str,
    end_date: str,
) -> Iterator[dict[str, Any]]:
    """Iterate aggregated summaries for a monitor across the given date range.

    Args:
        monitor_id: Monitor UUID.
        entry_type: Sensor field (e.g. ``'PM2.5'``).
        resolution: One of ``'hourly'``, ``'daily'``, ``'monthly'``,
            ``'quarterly'``, ``'seasonal'``, ``'yearly'``.
        start_date: ISO 8601 date string.
        end_date: ISO 8601 date string.
    """
    base = f'{self.PATH}{monitor_id}/summaries/'
    paths = _iter_summary_paths(
        base,
        entry_type,
        resolution,
        date.fromisoformat(start_date),
        date.fromisoformat(end_date),
    )
    rows = itertools.chain.from_iterable(self._paginate(p) for p in paths)
    # The API's summary rows don't identify their monitor; tag each row so
    # callers fanning out across monitors can attribute results.
    return ({'monitor_id': monitor_id, **row} for row in rows)

closest 

closest(entry_type, lat, lon)

Return up to 3 nearest active monitors with distance and latest entry.

Source code in sjvair/resources/monitors.py 
116
117
118
def closest(self, entry_type: str, lat: float, lon: float) -> list[dict[str, Any]]:  # ty: ignore[invalid-type-form]
    """Return up to 3 nearest active monitors with distance and latest entry."""
    return self._client.get(f'monitors/{entry_type}/closest/', {'lat': lat, 'lon': lon})['data']

current 

current(entry_type)

Iterate all active monitors with their most recent entry for the given type.

Source code in sjvair/resources/monitors.py 
120
121
122
def current(self, entry_type: str) -> Iterator[dict[str, Any]]:
    """Iterate all active monitors with their most recent entry for the given type."""
    return self._paginate(f'monitors/{entry_type}/current/')

sjvair.resources.regions.RegionsResource

Bases: BaseResource

Access geographic region data (counties, cities, ZIP codes, census tracts).

Available on :attr:SJVAirClient.regions.

Source code in sjvair/resources/regions.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class RegionsResource(BaseResource):
    """Access geographic region data (counties, cities, ZIP codes, census tracts).

    Available on :attr:`SJVAirClient.regions`.
    """

    PATH = 'regions/'

    def list(self, **params: Any) -> Iterator[dict[str, Any]]:
        """Iterate all regions, optionally filtered by ``kind``, ``county``, etc."""
        return self._paginate(self.PATH, params or None)

    def get(self, region_id: str) -> dict[str, Any]:
        """Get a single region by ID."""
        return self._client.get(f'{self.PATH}{region_id}/')['data']

    def search(self, query: str, **params: Any) -> list[dict[str, Any]]:  # ty: ignore[invalid-type-form]
        """Search regions by name, returning all high-confidence matches. Pass ``type=`` to scope to a specific region type."""
        return (
            self._client.get(
                f'{self.PATH}places/search/',
                {'q': query, **(params or {})},
            ).get('data')
            or []
        )

    def lookup(self, query: str, **params: Any) -> dict[str, Any] | None:
        """Resolve a name to the single best-match region. Pass ``type=`` to scope to a specific region type."""
        return self._client.get(
            f'{self.PATH}places/lookup/',
            {'q': query, **(params or {})},
        ).get('data')

    def summaries(
        self,
        region_id: str,
        entry_type: str,
        resolution: str,
        start_date: str,
        end_date: str,
    ) -> Iterator[dict[str, Any]]:
        """Iterate aggregated summaries for a region. Same resolution options as :meth:`MonitorsResource.summaries`."""
        base = f'{self.PATH}{region_id}/summaries/'
        paths = _iter_summary_paths(
            base,
            entry_type,
            resolution,
            date.fromisoformat(start_date),
            date.fromisoformat(end_date),
        )
        rows = itertools.chain.from_iterable(self._paginate(p) for p in paths)
        # The API's summary rows don't identify their region; tag each row so
        # callers can attribute results.
        return ({'region_id': region_id, **row} for row in rows)

list 

list(**params)

Iterate all regions, optionally filtered by kind, county, etc.

Source code in sjvair/resources/regions.py 
19
20
21
def list(self, **params: Any) -> Iterator[dict[str, Any]]:
    """Iterate all regions, optionally filtered by ``kind``, ``county``, etc."""
    return self._paginate(self.PATH, params or None)

get 

get(region_id)

Get a single region by ID.

Source code in sjvair/resources/regions.py 
23
24
25
def get(self, region_id: str) -> dict[str, Any]:
    """Get a single region by ID."""
    return self._client.get(f'{self.PATH}{region_id}/')['data']

search 

search(query, **params)

Search regions by name, returning all high-confidence matches. Pass type= to scope to a specific region type.

Source code in sjvair/resources/regions.py 
27
28
29
30
31
32
33
34
35
def search(self, query: str, **params: Any) -> list[dict[str, Any]]:  # ty: ignore[invalid-type-form]
    """Search regions by name, returning all high-confidence matches. Pass ``type=`` to scope to a specific region type."""
    return (
        self._client.get(
            f'{self.PATH}places/search/',
            {'q': query, **(params or {})},
        ).get('data')
        or []
    )

lookup 

lookup(query, **params)

Resolve a name to the single best-match region. Pass type= to scope to a specific region type.

Source code in sjvair/resources/regions.py 
37
38
39
40
41
42
def lookup(self, query: str, **params: Any) -> dict[str, Any] | None:
    """Resolve a name to the single best-match region. Pass ``type=`` to scope to a specific region type."""
    return self._client.get(
        f'{self.PATH}places/lookup/',
        {'q': query, **(params or {})},
    ).get('data')

summaries 

summaries(
    region_id, entry_type, resolution, start_date, end_date
)

Iterate aggregated summaries for a region. Same resolution options as :meth:MonitorsResource.summaries.

Source code in sjvair/resources/regions.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def summaries(
    self,
    region_id: str,
    entry_type: str,
    resolution: str,
    start_date: str,
    end_date: str,
) -> Iterator[dict[str, Any]]:
    """Iterate aggregated summaries for a region. Same resolution options as :meth:`MonitorsResource.summaries`."""
    base = f'{self.PATH}{region_id}/summaries/'
    paths = _iter_summary_paths(
        base,
        entry_type,
        resolution,
        date.fromisoformat(start_date),
        date.fromisoformat(end_date),
    )
    rows = itertools.chain.from_iterable(self._paginate(p) for p in paths)
    # The API's summary rows don't identify their region; tag each row so
    # callers can attribute results.
    return ({'region_id': region_id, **row} for row in rows)

sjvair.resources.calenviroscreen.CalEnviroScreenResource

Bases: BaseResource

Access CalEnviroScreen 4.0 census tract cumulative impact scores.

Available on :attr:SJVAirClient.calenviroscreen.

Source code in sjvair/resources/calenviroscreen.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
class CalEnviroScreenResource(BaseResource):
    """Access CalEnviroScreen 4.0 census tract cumulative impact scores.

    Available on :attr:`SJVAirClient.calenviroscreen`.
    """

    VERSION = '4.0'

    def list(self, year: int, **params: Any) -> Iterator[dict[str, Any]]:
        """Iterate all census tract scores for the given data year."""
        return self._paginate(f'calenviroscreen/{self.VERSION}/{year}/', params or None)

    def get(self, year: int, tract: str) -> dict[str, Any]:
        """Get CalEnviroScreen scores for a single census tract (FIPS code)."""
        return self._client.get(f'calenviroscreen/{self.VERSION}/{year}/{tract}/')['data']

list 

list(year, **params)

Iterate all census tract scores for the given data year.

Source code in sjvair/resources/calenviroscreen.py 
16
17
18
def list(self, year: int, **params: Any) -> Iterator[dict[str, Any]]:
    """Iterate all census tract scores for the given data year."""
    return self._paginate(f'calenviroscreen/{self.VERSION}/{year}/', params or None)

get 

get(year, tract)

Get CalEnviroScreen scores for a single census tract (FIPS code).

Source code in sjvair/resources/calenviroscreen.py 
20
21
22
def get(self, year: int, tract: str) -> dict[str, Any]:
    """Get CalEnviroScreen scores for a single census tract (FIPS code)."""
    return self._client.get(f'calenviroscreen/{self.VERSION}/{year}/{tract}/')['data']

sjvair.resources.ceidars.CEIDARSResource

Bases: BaseResource

Access CEIDARS (California Emissions Inventory) facility data.

Available on :attr:SJVAirClient.ceidars.

Source code in sjvair/resources/ceidars.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class CEIDARSResource(BaseResource):
    """Access CEIDARS (California Emissions Inventory) facility data.

    Available on :attr:`SJVAirClient.ceidars`.
    """

    PATH = 'ceidars/'

    def list(self, **params: Any) -> Iterator[dict[str, Any]]:
        """Iterate all CEIDARS facilities."""
        return self._paginate(self.PATH, params or None)

    def get(self, facility_id: str) -> dict[str, Any]:
        """Get a single CEIDARS facility by ID."""
        return self._client.get(f'{self.PATH}{facility_id}/')['data']

    def years(self) -> list[int]:  # ty: ignore[invalid-type-form]
        """Return the list of inventory years available in the dataset."""
        return self._client.get(f'{self.PATH}years/')['data']

list 

list(**params)

Iterate all CEIDARS facilities.

Source code in sjvair/resources/ceidars.py 
16
17
18
def list(self, **params: Any) -> Iterator[dict[str, Any]]:
    """Iterate all CEIDARS facilities."""
    return self._paginate(self.PATH, params or None)

get 

get(facility_id)

Get a single CEIDARS facility by ID.

Source code in sjvair/resources/ceidars.py 
20
21
22
def get(self, facility_id: str) -> dict[str, Any]:
    """Get a single CEIDARS facility by ID."""
    return self._client.get(f'{self.PATH}{facility_id}/')['data']

years 

years()

Return the list of inventory years available in the dataset.

Source code in sjvair/resources/ceidars.py 
24
25
26
def years(self) -> list[int]:  # ty: ignore[invalid-type-form]
    """Return the list of inventory years available in the dataset."""
    return self._client.get(f'{self.PATH}years/')['data']

sjvair.resources.hms.HMSResource

Bases: BaseResource

Access NOAA Hazard Mapping System (HMS) smoke and fire data.

Available on :attr:SJVAirClient.hms. Sub-resources:

  • :attr:smoke — smoke plume polygons
  • :attr:fire — fire detection points
Source code in sjvair/resources/hms.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
class HMSResource(BaseResource):
    """Access NOAA Hazard Mapping System (HMS) smoke and fire data.

    Available on :attr:`SJVAirClient.hms`. Sub-resources:

    - :attr:`smoke` — smoke plume polygons
    - :attr:`fire` — fire detection points
    """

    def __init__(self, client: Any) -> None:
        super().__init__(client)
        self.smoke = HMSSmokeResource(client)
        self.fire = HMSFireResource(client)

sjvair.resources.pesticides.PesticidesResource

Bases: BaseResource

Access California Department of Pesticide Regulation (CDPR) pesticide data.

Available on :attr:SJVAirClient.pesticides. Sub-resources:

  • :attr:chemicals — active ingredient lookup
  • :attr:commodities — crop/commodity lookup
  • :attr:products — registered product lookup
  • :attr:use — pesticide use reports
  • :attr:notice — pesticide use notices
Source code in sjvair/resources/pesticides.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
class PesticidesResource(BaseResource):
    """Access California Department of Pesticide Regulation (CDPR) pesticide data.

    Available on :attr:`SJVAirClient.pesticides`. Sub-resources:

    - :attr:`chemicals` — active ingredient lookup
    - :attr:`commodities` — crop/commodity lookup
    - :attr:`products` — registered product lookup
    - :attr:`use` — pesticide use reports
    - :attr:`notice` — pesticide use notices
    """

    def __init__(self, client: Any) -> None:
        super().__init__(client)
        self.chemicals = PesticidesChemicalsResource(client)
        self.commodities = PesticidesCommoditiesResource(client)
        self.products = PesticidesProductsResource(client)
        self.use = PesticidesUseResource(client)
        self.notice = PesticidesNoticeResource(client)

    def region_use(self, region_id: str, **params: Any) -> Iterator[dict[str, Any]]:
        """Iterate pesticide use reports for a specific region."""
        return self._paginate(f'pesticides/region/{region_id}/use/', params or None)

    def region_notice(self, region_id: str, **params: Any) -> Iterator[dict[str, Any]]:
        """Iterate pesticide use notices for a specific region."""
        return self._paginate(f'pesticides/region/{region_id}/notice/', params or None)

    def region_summary(self, region_id: str) -> dict[str, Any]:
        """Return an aggregate pesticide use summary for a region."""
        return self._client.get(f'pesticides/region/{region_id}/summary/')['data']

region_use 

region_use(region_id, **params)

Iterate pesticide use reports for a specific region.

Source code in sjvair/resources/pesticides.py 
70
71
72
def region_use(self, region_id: str, **params: Any) -> Iterator[dict[str, Any]]:
    """Iterate pesticide use reports for a specific region."""
    return self._paginate(f'pesticides/region/{region_id}/use/', params or None)

region_notice 

region_notice(region_id, **params)

Iterate pesticide use notices for a specific region.

Source code in sjvair/resources/pesticides.py 
74
75
76
def region_notice(self, region_id: str, **params: Any) -> Iterator[dict[str, Any]]:
    """Iterate pesticide use notices for a specific region."""
    return self._paginate(f'pesticides/region/{region_id}/notice/', params or None)

region_summary 

region_summary(region_id)

Return an aggregate pesticide use summary for a region.

Source code in sjvair/resources/pesticides.py 
78
79
80
def region_summary(self, region_id: str) -> dict[str, Any]:
    """Return an aggregate pesticide use summary for a region."""
    return self._client.get(f'pesticides/region/{region_id}/summary/')['data']