or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

client.mdindex.mdreferences.mdresources.mdsearch.mdutilities.md
tile.json

tessl/pypi-fhirpy

Async/sync FHIR client for Python providing comprehensive API for CRUD operations over FHIR resources

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/fhirpy@2.0.x

To install, run

npx @tessl/cli install tessl/pypi-fhirpy@2.0.0

index.mddocs/

fhirpy

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.

Package Information

  • Package Name: fhirpy
  • Package Type: pypi
  • Language: Python
  • Installation: pip install fhirpy
  • Python Version: 3.9+

Core Imports

from fhirpy import AsyncFHIRClient, SyncFHIRClient

For exceptions:

from fhirpy.base.exceptions import (
    ResourceNotFound, 
    InvalidResponse, 
    AuthorizationError, 
    MultipleResourcesFound,
    OperationOutcome
)

For search utilities:

from fhirpy.base.searchset import SQ, Raw

For data utilities:

from fhirpy.base.utils import AttrDict, SearchList

Basic Usage

Async Client

import asyncio
from fhirpy import AsyncFHIRClient

async def main():
    # Create client
    client = AsyncFHIRClient(
        'http://fhir-server/',
        authorization='Bearer TOKEN'
    )
    
    # Search for patients
    patients = client.resources('Patient').search(name='John').limit(10)
    results = await patients.fetch()
    
    # Create a new patient
    patient = client.resource('Patient', name=[{'text': 'John Doe'}])
    await patient.create()
    
    # Get patient by ID
    patient = await client.get('Patient', 'patient-id')
    
    # Update patient
    patient['active'] = True
    await patient.save()

asyncio.run(main())

Sync Client

from fhirpy import SyncFHIRClient

# Create client
client = SyncFHIRClient(
    'http://fhir-server/',
    authorization='Bearer TOKEN'
)

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

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

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

# Update patient
patient['active'] = True
patient.save()

Architecture

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

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

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

Capabilities

Client Operations

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

class AsyncFHIRClient:
    def __init__(
        self, 
        url: str, 
        authorization: str = None, 
        extra_headers: dict = None,
        aiohttp_config: dict = None,
        *,
        dump_resource: Callable = lambda x: dict(x)
    ): ...
    
    def resource(self, resource_type: str, **kwargs) -> AsyncFHIRResource: ...
    def resources(self, resource_type: str) -> AsyncFHIRSearchSet: ...
    def reference(
        self, 
        resource_type: str = None, 
        id: str = None, 
        reference: str = None, 
        **kwargs
    ) -> AsyncFHIRReference: ...
    
    async def get(self, resource_type: str, id: str) -> AsyncFHIRResource: ...
    async def create(self, resource, **kwargs): ...
    async def update(self, resource, **kwargs): ...
    async def patch(self, resource, **kwargs): ...
    async def delete(self, resource, **kwargs): ...
    async def save(self, resource, **kwargs): ...
    async def execute(
        self, 
        path: str, 
        method: str = "post", 
        data: dict = None, 
        params: dict = None
    ): ...

class SyncFHIRClient:
    def __init__(
        self, 
        url: str, 
        authorization: str = None, 
        extra_headers: dict = None,
        requests_config: dict = None,
        *,
        dump_resource: Callable = lambda x: dict(x)
    ): ...
    
    def resource(self, resource_type: str, **kwargs) -> SyncFHIRResource: ...
    def resources(self, resource_type: str) -> SyncFHIRSearchSet: ...
    def reference(
        self, 
        resource_type: str = None, 
        id: str = None, 
        reference: str = None, 
        **kwargs
    ) -> SyncFHIRReference: ...
    
    def get(self, resource_type: str, id: str) -> SyncFHIRResource: ...
    def create(self, resource, **kwargs): ...
    def update(self, resource, **kwargs): ...
    def patch(self, resource, **kwargs): ...
    def delete(self, resource, **kwargs): ...
    def save(self, resource, **kwargs): ...
    def execute(
        self, 
        path: str, 
        method: str = "post", 
        data: dict = None, 
        params: dict = None
    ): ...

Client Operations

Resource Management

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

class AsyncFHIRResource:
    async def create(self, **kwargs): ...
    async def update(self, **kwargs): ...
    async def patch(self, **kwargs): ...
    async def delete(self): ...
    async def save(self, **kwargs): ...
    async def refresh(self): ...
    async def is_valid(self) -> bool: ...
    
    def to_reference(self, **kwargs) -> AsyncFHIRReference: ...
    def serialize(self) -> dict: ...
    def get_by_path(self, path: str, default=None): ...
    
    @property
    def resource_type(self) -> str: ...
    @property
    def id(self) -> str: ...
    @property
    def reference(self) -> str: ...

class SyncFHIRResource:
    def create(self, **kwargs): ...
    def update(self, **kwargs): ...
    def patch(self, **kwargs): ...
    def delete(self): ...
    def save(self, **kwargs): ...
    def refresh(self): ...
    def is_valid(self) -> bool: ...
    
    def to_reference(self, **kwargs) -> SyncFHIRReference: ...
    def serialize(self) -> dict: ...
    def get_by_path(self, path: str, default=None): ...
    
    @property
    def resource_type(self) -> str: ...
    @property
    def id(self) -> str: ...
    @property
    def reference(self) -> str: ...

Resource Management

Search and Query

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

class AsyncFHIRSearchSet:
    def search(self, **params) -> AsyncFHIRSearchSet: ...
    def limit(self, count: int) -> AsyncFHIRSearchSet: ...
    def sort(self, *args) -> AsyncFHIRSearchSet: ...
    def elements(self, *args) -> AsyncFHIRSearchSet: ...
    def include(
        self, 
        resource_type: str, 
        attr: str = None, 
        *, 
        iterate: bool = False
    ) -> AsyncFHIRSearchSet: ...
    def revinclude(
        self, 
        resource_type: str, 
        attr: str = None, 
        *, 
        iterate: bool = False
    ) -> AsyncFHIRSearchSet: ...
    def has(self, *args, **kwargs) -> AsyncFHIRSearchSet: ...
    
    async def count(self) -> int: ...
    async def first(self) -> AsyncFHIRResource: ...
    async def get(self) -> AsyncFHIRResource: ...
    async def fetch(self) -> list: ...
    async def fetch_all(self) -> list: ...
    async def fetch_raw(self) -> dict: ...
    
    def clone(self) -> AsyncFHIRSearchSet: ...

class SyncFHIRSearchSet:
    def search(self, **params) -> SyncFHIRSearchSet: ...
    def limit(self, count: int) -> SyncFHIRSearchSet: ...
    def sort(self, *args) -> SyncFHIRSearchSet: ...
    def elements(self, *args) -> SyncFHIRSearchSet: ...
    def include(
        self, 
        resource_type: str, 
        attr: str = None, 
        *, 
        iterate: bool = False
    ) -> SyncFHIRSearchSet: ...
    def revinclude(
        self, 
        resource_type: str, 
        attr: str = None, 
        *, 
        iterate: bool = False
    ) -> SyncFHIRSearchSet: ...
    def has(self, *args, **kwargs) -> SyncFHIRSearchSet: ...
    
    def count(self) -> int: ...
    def first(self) -> SyncFHIRResource: ...
    def get(self) -> SyncFHIRResource: ...
    def fetch(self) -> list: ...
    def fetch_all(self) -> list: ...
    def fetch_raw(self) -> dict: ...
    
    def clone(self) -> SyncFHIRSearchSet: ...

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

class Raw:
    def __init__(self, **kwargs): ...

Search and Query

References

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

class AsyncFHIRReference:
    async def resolve(self) -> AsyncFHIRResource: ...
    
    @property
    def reference(self) -> str: ...
    @property
    def id(self) -> str: ...
    @property
    def resource_type(self) -> str: ...  
    @property
    def is_local(self) -> bool: ...

class SyncFHIRReference:
    def resolve(self) -> SyncFHIRResource: ...
    
    @property
    def reference(self) -> str: ...
    @property
    def id(self) -> str: ...
    @property
    def resource_type(self) -> str: ...
    @property
    def is_local(self) -> bool: ...

References

Utilities

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

class AttrDict(dict):
    def get_by_path(self, path: str, default=None): ...

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

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

def format_date_time(date: datetime) -> str: ...
def format_date(date: date) -> str: ...
def transform_param(param: str) -> str: ...
def transform_value(value) -> str: ...

Utilities

Exception Types

class BaseFHIRError(Exception):
    """Base exception for all FHIR errors"""

class ResourceNotFound(BaseFHIRError):
    """Resource not found error"""

class InvalidResponse(BaseFHIRError):  
    """Invalid response error"""

class AuthorizationError(BaseFHIRError):
    """Authorization error"""

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

class OperationOutcome(BaseFHIRError):
    """FHIR OperationOutcome error"""
    def __init__(
        self,
        reason=None,
        *,
        resource=None,
        severity="fatal",
        code="invalid"
    ): ...

class IssueType(Enum):
    invalid = "invalid"
    required = "required" 
    forbidden = "forbidden"
    not_found = "not-found"
    exception = "exception"
    informational = "informational"

class IssueSeverity(Enum):
    fatal = "fatal"
    error = "error"
    warning = "warning"
    information = "information"