Skip to content

Client API Reference

healthsites.client.HealthsitesClient

Async client for the Healthsites.io API v3.

Example

async with HealthsitesClient(api_key="your-api-key") as client: facilities = await client.list_facilities(country="ZA", page=1) print(facilities)

Source code in healthsites/client.py
 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
 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
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
class HealthsitesClient:
    """
    Async client for the Healthsites.io API v3.

    Example:
        async with HealthsitesClient(api_key="your-api-key") as client:
            facilities = await client.list_facilities(country="ZA", page=1)
            print(facilities)
    """

    BASE_URL = "https://healthsites.io/api/v3"

    def __init__(
        self,
        api_key: str,
        base_url: str | None = None,
        timeout: float = 30.0,
    ):
        """
        Initialize the Healthsites API client.

        Args:
            api_key: Your Healthsites API key.
            base_url: Optional custom base URL for the API.
            timeout: Request timeout in seconds (default: 30.0).
        """
        self.api_key = api_key
        self.base_url = base_url or self.BASE_URL
        self.timeout = timeout
        self._client: httpx.AsyncClient | None = None

    async def __aenter__(self) -> HealthsitesClient:
        """Enter async context manager."""
        self._client = httpx.AsyncClient(timeout=self.timeout)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        """Exit async context manager."""
        if self._client:
            await self._client.aclose()
            self._client = None

    @property
    def client(self) -> httpx.AsyncClient:
        """Get the HTTP client, creating one if necessary."""
        if self._client is None:
            self._client = httpx.AsyncClient(timeout=self.timeout)
        return self._client

    def _handle_response(self, response: httpx.Response) -> Any:
        """Handle API response and raise appropriate exceptions."""
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 401:
            raise AuthenticationError()
        elif response.status_code == 404:
            raise NotFoundError()
        elif response.status_code == 429:
            raise RateLimitError()
        elif response.status_code == 400:
            try:
                detail = response.json().get("detail", "Validation error")
            except Exception:
                detail = response.text
            raise ValidationError(str(detail))
        else:
            raise HealthsitesError(
                f"API error: {response.text}",
                status_code=response.status_code,
            )

    async def _get(
        self,
        endpoint: str,
        params: dict[str, Any] | None = None,
    ) -> Any:
        """Make a GET request to the API."""
        params = params or {}
        params["api-key"] = self.api_key
        # Remove None values
        params = {k: v for k, v in params.items() if v is not None}

        url = f"{self.base_url}{endpoint}"
        response = await self.client.get(url, params=params)
        return self._handle_response(response)

    async def _post(
        self,
        endpoint: str,
        data: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
    ) -> Any:
        """Make a POST request to the API."""
        params = params or {}
        params["api-key"] = self.api_key
        params = {k: v for k, v in params.items() if v is not None}

        url = f"{self.base_url}{endpoint}"
        response = await self.client.post(url, json=data, params=params)
        return self._handle_response(response)

    # -------------------------------------------------------------------------
    # Facilities Endpoints
    # -------------------------------------------------------------------------

    async def list_facilities(
        self,
        page: int = 1,
        country: str | None = None,
        extent: str | None = None,
        from_date: str | None = None,
        to_date: str | None = None,
        flat_properties: bool | None = None,
        tag_format: TagFormat | None = None,
        output: OutputFormat | None = None,
    ) -> dict[str, Any]:
        """
        List facilities with optional filtering.

        Args:
            page: Page number for pagination (required).
            country: Filter by country code (e.g., "ZA" for South Africa).
            extent: Bounding box as "min_lon,min_lat,max_lon,max_lat".
            from_date: Filter facilities updated from this date (ISO format).
            to_date: Filter facilities updated to this date (ISO format).
            flat_properties: Return flattened properties structure.
            tag_format: Tag format ("osm" or "hxl").
            output: Output format ("json", "geojson", or "xml").

        Returns:
            Dictionary containing facility data and pagination info.
        """
        params = {
            "page": page,
            "country": country,
            "extent": extent,
            "from": from_date,
            "to": to_date,
            "flat-properties": flat_properties,
            "tag-format": tag_format,
            "output": output,
        }
        return await self._get("/facilities/", params)

    async def create_facility(
        self,
        data: dict[str, Any],
    ) -> dict[str, Any]:
        """
        Create a new facility.

        Args:
            data: Facility data including geometry and properties.

        Returns:
            Created facility data.
        """
        return await self._post("/facilities/", data=data)

    async def get_facility(
        self,
        osm_type: OSMType,
        osm_id: int,
    ) -> dict[str, Any]:
        """
        Get a specific facility by OSM type and ID.

        Args:
            osm_type: OSM element type ("node", "way", or "relation").
            osm_id: OSM element ID.

        Returns:
            Facility detail data.
        """
        return await self._get(f"/facilities/{osm_type}/{osm_id}")

    async def update_facility(
        self,
        osm_type: OSMType,
        osm_id: int,
        data: dict[str, Any],
    ) -> dict[str, Any]:
        """
        Update an existing facility.

        Args:
            osm_type: OSM element type ("node", "way", or "relation").
            osm_id: OSM element ID.
            data: Updated facility data.

        Returns:
            Updated facility data.
        """
        return await self._post(f"/facilities/{osm_type}/{osm_id}", data=data)

    async def get_statistics(
        self,
        country: str | None = None,
        extent: str | None = None,
        from_date: str | None = None,
        to_date: str | None = None,
        flat_properties: bool | None = None,
        tag_format: TagFormat | None = None,
        output: OutputFormat | None = None,
    ) -> dict[str, Any]:
        """
        Get facility statistics.

        Args:
            country: Filter by country code (e.g., "ZA" for South Africa).
            extent: Bounding box as "min_lon,min_lat,max_lon,max_lat".
            from_date: Filter facilities updated from this date (ISO format).
            to_date: Filter facilities updated to this date (ISO format).
            flat_properties: Return flattened properties structure.
            tag_format: Tag format ("osm" or "hxl").
            output: Output format ("json", "geojson", or "xml").

        Returns:
            Statistics data for facilities.
        """
        params = {
            "country": country,
            "extent": extent,
            "from": from_date,
            "to": to_date,
            "flat-properties": flat_properties,
            "tag-format": tag_format,
            "output": output,
        }
        return await self._get("/facilities/statistic/", params)

    # -------------------------------------------------------------------------
    # Shapefile Endpoint
    # -------------------------------------------------------------------------

    async def download_shapefile(
        self,
        country: str,
        output_path: str | Path | None = None,
    ) -> bytes | Path:
        """
        Download shapefile data for a country.

        Args:
            country: Country code (e.g., "ZA" for South Africa).
            output_path: Optional path to save the shapefile. If not provided,
                        returns the raw bytes.

        Returns:
            Path to the saved file if output_path is provided, otherwise bytes.
        """
        url = f"{self.base_url}/shapefile/{country}"
        params = {"api-key": self.api_key}

        response = await self.client.get(url, params=params)

        if response.status_code != 200:
            self._handle_response(response)

        if output_path:
            output_path = Path(output_path)
            output_path.write_bytes(response.content)
            return output_path

        return response.content

    # -------------------------------------------------------------------------
    # User Endpoint
    # -------------------------------------------------------------------------

    async def get_user(self) -> dict[str, Any]:
        """
        Get the currently authenticated user's details.

        Returns:
            User detail data.
        """
        return await self._get("/user/")

    # -------------------------------------------------------------------------
    # Convenience Methods
    # -------------------------------------------------------------------------

    async def list_all_facilities(
        self,
        country: str | None = None,
        **kwargs,
    ) -> list[dict[str, Any]]:
        """
        Fetch all facilities across all pages.

        Args:
            country: Filter by country code.
            **kwargs: Additional filter parameters.

        Yields:
            All facilities matching the criteria.
        """
        all_facilities = []
        page = 1

        while True:
            result = await self.list_facilities(
                page=page,
                country=country,
                **kwargs,
            )
            facilities = result.get("features", result.get("results", []))
            if not facilities:
                break

            all_facilities.extend(facilities)
            page += 1

            # Check if we've reached the last page
            if "next" not in result or result.get("next") is None:
                break

        return all_facilities

    def list_endpoints(self) -> list[dict[str, str]]:
        """
        List all available API endpoints.

        Returns:
            List of endpoint information dictionaries.
        """
        return [
            {
                "method": "GET",
                "path": "/api/v3/facilities/",
                "description": "List facilities with filtering parameters",
                "function": "list_facilities()",
            },
            {
                "method": "POST",
                "path": "/api/v3/facilities/",
                "description": "Create a new facility",
                "function": "create_facility()",
            },
            {
                "method": "GET",
                "path": "/api/v3/facilities/statistic/",
                "description": "Get facility statistics",
                "function": "get_statistics()",
            },
            {
                "method": "GET",
                "path": "/api/v3/facilities/{osm_type}/{osm_id}",
                "description": "Get a specific facility by OSM type and ID",
                "function": "get_facility()",
            },
            {
                "method": "POST",
                "path": "/api/v3/facilities/{osm_type}/{osm_id}",
                "description": "Update an existing facility",
                "function": "update_facility()",
            },
            {
                "method": "GET",
                "path": "/api/v3/shapefile/{country}",
                "description": "Download shapefile data for a country",
                "function": "download_shapefile()",
            },
            {
                "method": "GET",
                "path": "/api/v3/user/",
                "description": "Get current user details",
                "function": "get_user()",
            },
        ]

    async def close(self) -> None:
        """Close the HTTP client."""
        if self._client:
            await self._client.aclose()
            self._client = None

Attributes

client property

Get the HTTP client, creating one if necessary.

Functions

__aenter__() async

Enter async context manager.

Source code in healthsites/client.py
59
60
61
62
async def __aenter__(self) -> HealthsitesClient:
    """Enter async context manager."""
    self._client = httpx.AsyncClient(timeout=self.timeout)
    return self

__aexit__(exc_type, exc_val, exc_tb) async

Exit async context manager.

Source code in healthsites/client.py
64
65
66
67
68
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
    """Exit async context manager."""
    if self._client:
        await self._client.aclose()
        self._client = None

__init__(api_key, base_url=None, timeout=30.0)

Initialize the Healthsites API client.

Parameters:

Name Type Description Default
api_key str

Your Healthsites API key.

required
base_url str | None

Optional custom base URL for the API.

None
timeout float

Request timeout in seconds (default: 30.0).

30.0
Source code in healthsites/client.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def __init__(
    self,
    api_key: str,
    base_url: str | None = None,
    timeout: float = 30.0,
):
    """
    Initialize the Healthsites API client.

    Args:
        api_key: Your Healthsites API key.
        base_url: Optional custom base URL for the API.
        timeout: Request timeout in seconds (default: 30.0).
    """
    self.api_key = api_key
    self.base_url = base_url or self.BASE_URL
    self.timeout = timeout
    self._client: httpx.AsyncClient | None = None

close() async

Close the HTTP client.

Source code in healthsites/client.py
400
401
402
403
404
async def close(self) -> None:
    """Close the HTTP client."""
    if self._client:
        await self._client.aclose()
        self._client = None

create_facility(data) async

Create a new facility.

Parameters:

Name Type Description Default
data dict[str, Any]

Facility data including geometry and properties.

required

Returns:

Type Description
dict[str, Any]

Created facility data.

Source code in healthsites/client.py
172
173
174
175
176
177
178
179
180
181
182
183
184
185
async def create_facility(
    self,
    data: dict[str, Any],
) -> dict[str, Any]:
    """
    Create a new facility.

    Args:
        data: Facility data including geometry and properties.

    Returns:
        Created facility data.
    """
    return await self._post("/facilities/", data=data)

download_shapefile(country, output_path=None) async

Download shapefile data for a country.

Parameters:

Name Type Description Default
country str

Country code (e.g., "ZA" for South Africa).

required
output_path str | Path | None

Optional path to save the shapefile. If not provided, returns the raw bytes.

None

Returns:

Type Description
bytes | Path

Path to the saved file if output_path is provided, otherwise bytes.

Source code in healthsites/client.py
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
async def download_shapefile(
    self,
    country: str,
    output_path: str | Path | None = None,
) -> bytes | Path:
    """
    Download shapefile data for a country.

    Args:
        country: Country code (e.g., "ZA" for South Africa).
        output_path: Optional path to save the shapefile. If not provided,
                    returns the raw bytes.

    Returns:
        Path to the saved file if output_path is provided, otherwise bytes.
    """
    url = f"{self.base_url}/shapefile/{country}"
    params = {"api-key": self.api_key}

    response = await self.client.get(url, params=params)

    if response.status_code != 200:
        self._handle_response(response)

    if output_path:
        output_path = Path(output_path)
        output_path.write_bytes(response.content)
        return output_path

    return response.content

get_facility(osm_type, osm_id) async

Get a specific facility by OSM type and ID.

Parameters:

Name Type Description Default
osm_type OSMType

OSM element type ("node", "way", or "relation").

required
osm_id int

OSM element ID.

required

Returns:

Type Description
dict[str, Any]

Facility detail data.

Source code in healthsites/client.py
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
async def get_facility(
    self,
    osm_type: OSMType,
    osm_id: int,
) -> dict[str, Any]:
    """
    Get a specific facility by OSM type and ID.

    Args:
        osm_type: OSM element type ("node", "way", or "relation").
        osm_id: OSM element ID.

    Returns:
        Facility detail data.
    """
    return await self._get(f"/facilities/{osm_type}/{osm_id}")

get_statistics(country=None, extent=None, from_date=None, to_date=None, flat_properties=None, tag_format=None, output=None) async

Get facility statistics.

Parameters:

Name Type Description Default
country str | None

Filter by country code (e.g., "ZA" for South Africa).

None
extent str | None

Bounding box as "min_lon,min_lat,max_lon,max_lat".

None
from_date str | None

Filter facilities updated from this date (ISO format).

None
to_date str | None

Filter facilities updated to this date (ISO format).

None
flat_properties bool | None

Return flattened properties structure.

None
tag_format TagFormat | None

Tag format ("osm" or "hxl").

None
output OutputFormat | None

Output format ("json", "geojson", or "xml").

None

Returns:

Type Description
dict[str, Any]

Statistics data for facilities.

Source code in healthsites/client.py
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
async def get_statistics(
    self,
    country: str | None = None,
    extent: str | None = None,
    from_date: str | None = None,
    to_date: str | None = None,
    flat_properties: bool | None = None,
    tag_format: TagFormat | None = None,
    output: OutputFormat | None = None,
) -> dict[str, Any]:
    """
    Get facility statistics.

    Args:
        country: Filter by country code (e.g., "ZA" for South Africa).
        extent: Bounding box as "min_lon,min_lat,max_lon,max_lat".
        from_date: Filter facilities updated from this date (ISO format).
        to_date: Filter facilities updated to this date (ISO format).
        flat_properties: Return flattened properties structure.
        tag_format: Tag format ("osm" or "hxl").
        output: Output format ("json", "geojson", or "xml").

    Returns:
        Statistics data for facilities.
    """
    params = {
        "country": country,
        "extent": extent,
        "from": from_date,
        "to": to_date,
        "flat-properties": flat_properties,
        "tag-format": tag_format,
        "output": output,
    }
    return await self._get("/facilities/statistic/", params)

get_user() async

Get the currently authenticated user's details.

Returns:

Type Description
dict[str, Any]

User detail data.

Source code in healthsites/client.py
298
299
300
301
302
303
304
305
async def get_user(self) -> dict[str, Any]:
    """
    Get the currently authenticated user's details.

    Returns:
        User detail data.
    """
    return await self._get("/user/")

list_all_facilities(country=None, **kwargs) async

Fetch all facilities across all pages.

Parameters:

Name Type Description Default
country str | None

Filter by country code.

None
**kwargs

Additional filter parameters.

{}

Yields:

Type Description
list[dict[str, Any]]

All facilities matching the criteria.

Source code in healthsites/client.py
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
async def list_all_facilities(
    self,
    country: str | None = None,
    **kwargs,
) -> list[dict[str, Any]]:
    """
    Fetch all facilities across all pages.

    Args:
        country: Filter by country code.
        **kwargs: Additional filter parameters.

    Yields:
        All facilities matching the criteria.
    """
    all_facilities = []
    page = 1

    while True:
        result = await self.list_facilities(
            page=page,
            country=country,
            **kwargs,
        )
        facilities = result.get("features", result.get("results", []))
        if not facilities:
            break

        all_facilities.extend(facilities)
        page += 1

        # Check if we've reached the last page
        if "next" not in result or result.get("next") is None:
            break

    return all_facilities

list_endpoints()

List all available API endpoints.

Returns:

Type Description
list[dict[str, str]]

List of endpoint information dictionaries.

Source code in healthsites/client.py
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
def list_endpoints(self) -> list[dict[str, str]]:
    """
    List all available API endpoints.

    Returns:
        List of endpoint information dictionaries.
    """
    return [
        {
            "method": "GET",
            "path": "/api/v3/facilities/",
            "description": "List facilities with filtering parameters",
            "function": "list_facilities()",
        },
        {
            "method": "POST",
            "path": "/api/v3/facilities/",
            "description": "Create a new facility",
            "function": "create_facility()",
        },
        {
            "method": "GET",
            "path": "/api/v3/facilities/statistic/",
            "description": "Get facility statistics",
            "function": "get_statistics()",
        },
        {
            "method": "GET",
            "path": "/api/v3/facilities/{osm_type}/{osm_id}",
            "description": "Get a specific facility by OSM type and ID",
            "function": "get_facility()",
        },
        {
            "method": "POST",
            "path": "/api/v3/facilities/{osm_type}/{osm_id}",
            "description": "Update an existing facility",
            "function": "update_facility()",
        },
        {
            "method": "GET",
            "path": "/api/v3/shapefile/{country}",
            "description": "Download shapefile data for a country",
            "function": "download_shapefile()",
        },
        {
            "method": "GET",
            "path": "/api/v3/user/",
            "description": "Get current user details",
            "function": "get_user()",
        },
    ]

list_facilities(page=1, country=None, extent=None, from_date=None, to_date=None, flat_properties=None, tag_format=None, output=None) async

List facilities with optional filtering.

Parameters:

Name Type Description Default
page int

Page number for pagination (required).

1
country str | None

Filter by country code (e.g., "ZA" for South Africa).

None
extent str | None

Bounding box as "min_lon,min_lat,max_lon,max_lat".

None
from_date str | None

Filter facilities updated from this date (ISO format).

None
to_date str | None

Filter facilities updated to this date (ISO format).

None
flat_properties bool | None

Return flattened properties structure.

None
tag_format TagFormat | None

Tag format ("osm" or "hxl").

None
output OutputFormat | None

Output format ("json", "geojson", or "xml").

None

Returns:

Type Description
dict[str, Any]

Dictionary containing facility data and pagination info.

Source code in healthsites/client.py
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
async def list_facilities(
    self,
    page: int = 1,
    country: str | None = None,
    extent: str | None = None,
    from_date: str | None = None,
    to_date: str | None = None,
    flat_properties: bool | None = None,
    tag_format: TagFormat | None = None,
    output: OutputFormat | None = None,
) -> dict[str, Any]:
    """
    List facilities with optional filtering.

    Args:
        page: Page number for pagination (required).
        country: Filter by country code (e.g., "ZA" for South Africa).
        extent: Bounding box as "min_lon,min_lat,max_lon,max_lat".
        from_date: Filter facilities updated from this date (ISO format).
        to_date: Filter facilities updated to this date (ISO format).
        flat_properties: Return flattened properties structure.
        tag_format: Tag format ("osm" or "hxl").
        output: Output format ("json", "geojson", or "xml").

    Returns:
        Dictionary containing facility data and pagination info.
    """
    params = {
        "page": page,
        "country": country,
        "extent": extent,
        "from": from_date,
        "to": to_date,
        "flat-properties": flat_properties,
        "tag-format": tag_format,
        "output": output,
    }
    return await self._get("/facilities/", params)

update_facility(osm_type, osm_id, data) async

Update an existing facility.

Parameters:

Name Type Description Default
osm_type OSMType

OSM element type ("node", "way", or "relation").

required
osm_id int

OSM element ID.

required
data dict[str, Any]

Updated facility data.

required

Returns:

Type Description
dict[str, Any]

Updated facility data.

Source code in healthsites/client.py
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
async def update_facility(
    self,
    osm_type: OSMType,
    osm_id: int,
    data: dict[str, Any],
) -> dict[str, Any]:
    """
    Update an existing facility.

    Args:
        osm_type: OSM element type ("node", "way", or "relation").
        osm_id: OSM element ID.
        data: Updated facility data.

    Returns:
        Updated facility data.
    """
    return await self._post(f"/facilities/{osm_type}/{osm_id}", data=data)

healthsites.client.HealthsitesClientSync

Synchronous wrapper for HealthsitesClient.

Example

client = HealthsitesClientSync(api_key="your-api-key") facilities = client.list_facilities(country="ZA", page=1) print(facilities) client.close()

Source code in healthsites/client.py
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
class HealthsitesClientSync:
    """
    Synchronous wrapper for HealthsitesClient.

    Example:
        client = HealthsitesClientSync(api_key="your-api-key")
        facilities = client.list_facilities(country="ZA", page=1)
        print(facilities)
        client.close()
    """

    def __init__(self, api_key: str, **kwargs):
        """Initialize synchronous client."""
        self._async_client = HealthsitesClient(api_key, **kwargs)

    def _run(self, coro):
        """Run async coroutine synchronously."""
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        return loop.run_until_complete(coro)

    def list_facilities(self, **kwargs):
        """List facilities (sync wrapper)."""
        return self._run(self._async_client.list_facilities(**kwargs))

    def create_facility(self, data):
        """Create facility (sync wrapper)."""
        return self._run(self._async_client.create_facility(data))

    def get_facility(self, osm_type, osm_id):
        """Get facility (sync wrapper)."""
        return self._run(self._async_client.get_facility(osm_type, osm_id))

    def update_facility(self, osm_type, osm_id, data):
        """Update facility (sync wrapper)."""
        return self._run(self._async_client.update_facility(osm_type, osm_id, data))

    def get_statistics(self, **kwargs):
        """Get statistics (sync wrapper)."""
        return self._run(self._async_client.get_statistics(**kwargs))

    def download_shapefile(self, country, output_path=None):
        """Download shapefile (sync wrapper)."""
        return self._run(self._async_client.download_shapefile(country, output_path))

    def get_user(self):
        """Get user (sync wrapper)."""
        return self._run(self._async_client.get_user())

    def list_all_facilities(self, **kwargs):
        """List all facilities (sync wrapper)."""
        return self._run(self._async_client.list_all_facilities(**kwargs))

    def list_endpoints(self):
        """List all endpoints."""
        return self._async_client.list_endpoints()

    def close(self):
        """Close the client."""
        self._run(self._async_client.close())

Functions

__init__(api_key, **kwargs)

Initialize synchronous client.

Source code in healthsites/client.py
419
420
421
def __init__(self, api_key: str, **kwargs):
    """Initialize synchronous client."""
    self._async_client = HealthsitesClient(api_key, **kwargs)

close()

Close the client.

Source code in healthsites/client.py
468
469
470
def close(self):
    """Close the client."""
    self._run(self._async_client.close())

create_facility(data)

Create facility (sync wrapper).

Source code in healthsites/client.py
436
437
438
def create_facility(self, data):
    """Create facility (sync wrapper)."""
    return self._run(self._async_client.create_facility(data))

download_shapefile(country, output_path=None)

Download shapefile (sync wrapper).

Source code in healthsites/client.py
452
453
454
def download_shapefile(self, country, output_path=None):
    """Download shapefile (sync wrapper)."""
    return self._run(self._async_client.download_shapefile(country, output_path))

get_facility(osm_type, osm_id)

Get facility (sync wrapper).

Source code in healthsites/client.py
440
441
442
def get_facility(self, osm_type, osm_id):
    """Get facility (sync wrapper)."""
    return self._run(self._async_client.get_facility(osm_type, osm_id))

get_statistics(**kwargs)

Get statistics (sync wrapper).

Source code in healthsites/client.py
448
449
450
def get_statistics(self, **kwargs):
    """Get statistics (sync wrapper)."""
    return self._run(self._async_client.get_statistics(**kwargs))

get_user()

Get user (sync wrapper).

Source code in healthsites/client.py
456
457
458
def get_user(self):
    """Get user (sync wrapper)."""
    return self._run(self._async_client.get_user())

list_all_facilities(**kwargs)

List all facilities (sync wrapper).

Source code in healthsites/client.py
460
461
462
def list_all_facilities(self, **kwargs):
    """List all facilities (sync wrapper)."""
    return self._run(self._async_client.list_all_facilities(**kwargs))

list_endpoints()

List all endpoints.

Source code in healthsites/client.py
464
465
466
def list_endpoints(self):
    """List all endpoints."""
    return self._async_client.list_endpoints()

list_facilities(**kwargs)

List facilities (sync wrapper).

Source code in healthsites/client.py
432
433
434
def list_facilities(self, **kwargs):
    """List facilities (sync wrapper)."""
    return self._run(self._async_client.list_facilities(**kwargs))

update_facility(osm_type, osm_id, data)

Update facility (sync wrapper).

Source code in healthsites/client.py
444
445
446
def update_facility(self, osm_type, osm_id, data):
    """Update facility (sync wrapper)."""
    return self._run(self._async_client.update_facility(osm_type, osm_id, data))

Made with love by Kartoza | Donate! | GitHub