Skip to content

apple_health

apple_health

Apple Health connector -- reads HealthKit SQLite DB or iPhone Health export XML.

Two data sources are tried in order: 1. HealthKit SQLite DB at ~/Library/Health/healthdb_secure.sqlite (macOS with HealthKit sync enabled). 2. Health Export XML placed by the user at ~/.openjarvis/connectors/apple_health_export/export.xml.

Both are local-only; no API keys are needed.

Classes

AppleHealthConnector 

AppleHealthConnector(*, export_path: str = _DEFAULT_EXPORT_PATH, healthkit_db_path: str = _DEFAULT_HEALTHKIT_DB_PATH)

Bases: BaseConnector

Sync health data from Apple Health (local files only).

Source code in src/openjarvis/connectors/apple_health.py 
82
83
84
85
86
87
88
89
90
def __init__(
    self,
    *,
    export_path: str = _DEFAULT_EXPORT_PATH,
    healthkit_db_path: str = _DEFAULT_HEALTHKIT_DB_PATH,
) -> None:
    self._export_path = Path(export_path)
    self._healthkit_db_path = Path(healthkit_db_path)
    self._status = SyncStatus()
Functions
sync 
sync(*, since: Optional[datetime] = None, cursor: Optional[str] = None) -> Iterator[Document]

Yield health Documents, preferring HealthKit DB over export XML.

Source code in src/openjarvis/connectors/apple_health.py 
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
def sync(
    self, *, since: Optional[datetime] = None, cursor: Optional[str] = None
) -> Iterator[Document]:
    """Yield health Documents, preferring HealthKit DB over export XML."""
    self._status.state = "syncing"
    try:
        # Try HealthKit SQLite first
        if self._healthkit_db_path.exists():
            try:
                yield from self._sync_healthkit_db(since=since)
                self._status.state = "idle"
                self._status.last_sync = datetime.now()
                return
            except Exception:
                logger.debug(
                    "HealthKit DB not readable, falling back to export XML",
                    exc_info=True,
                )

        # Fall back to export XML
        if self._export_path.exists():
            yield from self._sync_export_xml(since=since)

        self._status.state = "idle"
        self._status.last_sync = datetime.now()
    except Exception as exc:
        self._status.state = "error"
        self._status.error = str(exc)
        raise