0
# fhirpy
1

2
Async/sync FHIR client for Python providing comprehensive API for CRUD operations over FHIR resources. This package offers both asynchronous (based on aiohttp) and synchronous (based on requests) clients for maximum flexibility in healthcare applications requiring FHIR server interactions.
3

4
## Package Information
5

6
- **Package Name**: fhirpy
7
- **Package Type**: pypi
8
- **Language**: Python
9
- **Installation**: `pip install fhirpy`
10
- **Python Version**: 3.9+
11

12
## Core Imports
13

14
```python
15
from fhirpy import AsyncFHIRClient, SyncFHIRClient
16
```
17

18
For exceptions:
19

20
```python
21
from fhirpy.base.exceptions import (
22
    ResourceNotFound, 
23
    InvalidResponse, 
24
    AuthorizationError, 
25
    MultipleResourcesFound,
26
    OperationOutcome
27
)
28
```
29

30
For search utilities:
31

32
```python
33
from fhirpy.base.searchset import SQ, Raw
34
```
35

36
For data utilities:
37

38
```python
39
from fhirpy.base.utils import AttrDict, SearchList
40
```
41

42
## Basic Usage
43

44
### Async Client
45

46
```python
47
import asyncio
48
from fhirpy import AsyncFHIRClient
49

50
async def main():
51
    # Create client
52
    client = AsyncFHIRClient(
53
        'http://fhir-server/',
54
        authorization='Bearer TOKEN'
55
    )
56
    
57
    # Search for patients
58
    patients = client.resources('Patient').search(name='John').limit(10)
59
    results = await patients.fetch()
60
    
61
    # Create a new patient
62
    patient = client.resource('Patient', name=[{'text': 'John Doe'}])
63
    await patient.create()
64
    
65
    # Get patient by ID
66
    patient = await client.get('Patient', 'patient-id')
67
    
68
    # Update patient
69
    patient['active'] = True
70
    await patient.save()
71

72
asyncio.run(main())
73
```
74

75
### Sync Client
76

77
```python
78
from fhirpy import SyncFHIRClient
79

80
# Create client
81
client = SyncFHIRClient(
82
    'http://fhir-server/',
83
    authorization='Bearer TOKEN'
84
)
85

86
# Search for patients
87
patients = client.resources('Patient').search(name='John').limit(10)
88
results = patients.fetch()
89

90
# Create a new patient
91
patient = client.resource('Patient', name=[{'text': 'John Doe'}])
92
patient.create()
93

94
# Get patient by ID
95
patient = client.get('Patient', 'patient-id')
96

97
# Update patient
98
patient['active'] = True
99
patient.save()
100
```
101

102
## Architecture
103

104
The fhirpy library is built around a clear separation between async and sync implementations, sharing common base classes and protocols:
105

106
- **Client Layer**: AsyncFHIRClient and SyncFHIRClient provide the main entry points for FHIR server interaction
107
- **Resource Layer**: FHIRResource classes represent individual FHIR resources with full CRUD capabilities
108
- **SearchSet Layer**: FHIRSearchSet classes provide advanced query building and result fetching
109
- **Reference Layer**: FHIRReference classes handle FHIR resource references and resolution
110
- **Base Layer**: Abstract base classes provide common functionality shared between async/sync implementations
111

112
This design ensures consistent APIs across both async and sync modes while maintaining maximum flexibility for different use cases.
113

114
## Capabilities
115

116
### Client Operations
117

118
Core client functionality for connecting to FHIR servers, creating resources and search sets, and performing direct CRUD operations.
119

120
```python { .api }
121
class AsyncFHIRClient:
122
    def __init__(
123
        self, 
124
        url: str, 
125
        authorization: str = None, 
126
        extra_headers: dict = None,
127
        aiohttp_config: dict = None,
128
        *,
129
        dump_resource: Callable = lambda x: dict(x)
130
    ): ...
131
    
132
    def resource(self, resource_type: str, **kwargs) -> AsyncFHIRResource: ...
133
    def resources(self, resource_type: str) -> AsyncFHIRSearchSet: ...
134
    def reference(
135
        self, 
136
        resource_type: str = None, 
137
        id: str = None, 
138
        reference: str = None, 
139
        **kwargs
140
    ) -> AsyncFHIRReference: ...
141
    
142
    async def get(self, resource_type: str, id: str) -> AsyncFHIRResource: ...
143
    async def create(self, resource, **kwargs): ...
144
    async def update(self, resource, **kwargs): ...
145
    async def patch(self, resource, **kwargs): ...
146
    async def delete(self, resource, **kwargs): ...
147
    async def save(self, resource, **kwargs): ...
148
    async def execute(
149
        self, 
150
        path: str, 
151
        method: str = "post", 
152
        data: dict = None, 
153
        params: dict = None
154
    ): ...
155

156
class SyncFHIRClient:
157
    def __init__(
158
        self, 
159
        url: str, 
160
        authorization: str = None, 
161
        extra_headers: dict = None,
162
        requests_config: dict = None,
163
        *,
164
        dump_resource: Callable = lambda x: dict(x)
165
    ): ...
166
    
167
    def resource(self, resource_type: str, **kwargs) -> SyncFHIRResource: ...
168
    def resources(self, resource_type: str) -> SyncFHIRSearchSet: ...
169
    def reference(
170
        self, 
171
        resource_type: str = None, 
172
        id: str = None, 
173
        reference: str = None, 
174
        **kwargs
175
    ) -> SyncFHIRReference: ...
176
    
177
    def get(self, resource_type: str, id: str) -> SyncFHIRResource: ...
178
    def create(self, resource, **kwargs): ...
179
    def update(self, resource, **kwargs): ...
180
    def patch(self, resource, **kwargs): ...
181
    def delete(self, resource, **kwargs): ...
182
    def save(self, resource, **kwargs): ...
183
    def execute(
184
        self, 
185
        path: str, 
186
        method: str = "post", 
187
        data: dict = None, 
188
        params: dict = None
189
    ): ...
190
```
191

192
[Client Operations](./client.md)
193

194
### Resource Management
195

196
FHIR resource instances with full CRUD capabilities, validation, serialization, and path-based data access.
197

198
```python { .api }
199
class AsyncFHIRResource:
200
    async def create(self, **kwargs): ...
201
    async def update(self, **kwargs): ...
202
    async def patch(self, **kwargs): ...
203
    async def delete(self): ...
204
    async def save(self, **kwargs): ...
205
    async def refresh(self): ...
206
    async def is_valid(self) -> bool: ...
207
    
208
    def to_reference(self, **kwargs) -> AsyncFHIRReference: ...
209
    def serialize(self) -> dict: ...
210
    def get_by_path(self, path: str, default=None): ...
211
    
212
    @property
213
    def resource_type(self) -> str: ...
214
    @property
215
    def id(self) -> str: ...
216
    @property
217
    def reference(self) -> str: ...
218

219
class SyncFHIRResource:
220
    def create(self, **kwargs): ...
221
    def update(self, **kwargs): ...
222
    def patch(self, **kwargs): ...
223
    def delete(self): ...
224
    def save(self, **kwargs): ...
225
    def refresh(self): ...
226
    def is_valid(self) -> bool: ...
227
    
228
    def to_reference(self, **kwargs) -> SyncFHIRReference: ...
229
    def serialize(self) -> dict: ...
230
    def get_by_path(self, path: str, default=None): ...
231
    
232
    @property
233
    def resource_type(self) -> str: ...
234
    @property
235
    def id(self) -> str: ...
236
    @property
237
    def reference(self) -> str: ...
238
```
239

240
[Resource Management](./resources.md)
241

242
### Search and Query
243

244
Advanced FHIR search capabilities with chaining, modifiers, includes, pagination, and complex query building.
245

246
```python { .api }
247
class AsyncFHIRSearchSet:
248
    def search(self, **params) -> AsyncFHIRSearchSet: ...
249
    def limit(self, count: int) -> AsyncFHIRSearchSet: ...
250
    def sort(self, *args) -> AsyncFHIRSearchSet: ...
251
    def elements(self, *args) -> AsyncFHIRSearchSet: ...
252
    def include(
253
        self, 
254
        resource_type: str, 
255
        attr: str = None, 
256
        *, 
257
        iterate: bool = False
258
    ) -> AsyncFHIRSearchSet: ...
259
    def revinclude(
260
        self, 
261
        resource_type: str, 
262
        attr: str = None, 
263
        *, 
264
        iterate: bool = False
265
    ) -> AsyncFHIRSearchSet: ...
266
    def has(self, *args, **kwargs) -> AsyncFHIRSearchSet: ...
267
    
268
    async def count(self) -> int: ...
269
    async def first(self) -> AsyncFHIRResource: ...
270
    async def get(self) -> AsyncFHIRResource: ...
271
    async def fetch(self) -> list: ...
272
    async def fetch_all(self) -> list: ...
273
    async def fetch_raw(self) -> dict: ...
274
    
275
    def clone(self) -> AsyncFHIRSearchSet: ...
276

277
class SyncFHIRSearchSet:
278
    def search(self, **params) -> SyncFHIRSearchSet: ...
279
    def limit(self, count: int) -> SyncFHIRSearchSet: ...
280
    def sort(self, *args) -> SyncFHIRSearchSet: ...
281
    def elements(self, *args) -> SyncFHIRSearchSet: ...
282
    def include(
283
        self, 
284
        resource_type: str, 
285
        attr: str = None, 
286
        *, 
287
        iterate: bool = False
288
    ) -> SyncFHIRSearchSet: ...
289
    def revinclude(
290
        self, 
291
        resource_type: str, 
292
        attr: str = None, 
293
        *, 
294
        iterate: bool = False
295
    ) -> SyncFHIRSearchSet: ...
296
    def has(self, *args, **kwargs) -> SyncFHIRSearchSet: ...
297
    
298
    def count(self) -> int: ...
299
    def first(self) -> SyncFHIRResource: ...
300
    def get(self) -> SyncFHIRResource: ...
301
    def fetch(self) -> list: ...
302
    def fetch_all(self) -> list: ...
303
    def fetch_raw(self) -> dict: ...
304
    
305
    def clone(self) -> SyncFHIRSearchSet: ...
306

307
def SQ(*args, **kwargs) -> dict:
308
    """Build advanced search query parameters"""
309

310
class Raw:
311
    def __init__(self, **kwargs): ...
312
```
313

314
[Search and Query](./search.md)
315

316
### References
317

318
FHIR reference handling with resolution capabilities and local/external reference detection.
319

320
```python { .api }
321
class AsyncFHIRReference:
322
    async def resolve(self) -> AsyncFHIRResource: ...
323
    
324
    @property
325
    def reference(self) -> str: ...
326
    @property
327
    def id(self) -> str: ...
328
    @property
329
    def resource_type(self) -> str: ...  
330
    @property
331
    def is_local(self) -> bool: ...
332

333
class SyncFHIRReference:
334
    def resolve(self) -> SyncFHIRResource: ...
335
    
336
    @property
337
    def reference(self) -> str: ...
338
    @property
339
    def id(self) -> str: ...
340
    @property
341
    def resource_type(self) -> str: ...
342
    @property
343
    def is_local(self) -> bool: ...
344
```
345

346
[References](./references.md)
347

348
### Utilities
349

350
Helper functions and classes for data manipulation, parameter transformation, and path-based access.
351

352
```python { .api }
353
class AttrDict(dict):
354
    def get_by_path(self, path: str, default=None): ...
355

356
class SearchList(list):
357
    def get_by_path(self, path: str, default=None): ...
358

359
def chunks(lst: list, n: int) -> Generator: ...
360
def unique_everseen(seq: list) -> list: ...
361
def encode_params(params: dict) -> str: ...
362
def parse_path(path: str) -> list: ...
363
def get_by_path(obj, path: list, default=None): ...
364
def set_by_path(obj, path: str, value): ...
365

366
def format_date_time(date: datetime) -> str: ...
367
def format_date(date: date) -> str: ...
368
def transform_param(param: str) -> str: ...
369
def transform_value(value) -> str: ...
370
```
371

372
[Utilities](./utilities.md)
373

374
## Exception Types
375

376
```python { .api }
377
class BaseFHIRError(Exception):
378
    """Base exception for all FHIR errors"""
379

380
class ResourceNotFound(BaseFHIRError):
381
    """Resource not found error"""
382

383
class InvalidResponse(BaseFHIRError):  
384
    """Invalid response error"""
385

386
class AuthorizationError(BaseFHIRError):
387
    """Authorization error"""
388

389
class MultipleResourcesFound(BaseFHIRError):
390
    """Multiple resources found when expecting one"""
391

392
class OperationOutcome(BaseFHIRError):
393
    """FHIR OperationOutcome error"""
394
    def __init__(
395
        self,
396
        reason=None,
397
        *,
398
        resource=None,
399
        severity="fatal",
400
        code="invalid"
401
    ): ...
402

403
class IssueType(Enum):
404
    invalid = "invalid"
405
    required = "required" 
406
    forbidden = "forbidden"
407
    not_found = "not-found"
408
    exception = "exception"
409
    informational = "informational"
410

411
class IssueSeverity(Enum):
412
    fatal = "fatal"
413
    error = "error"
414
    warning = "warning"
415
    information = "information"
416
```