Using syncfm-core in Python

syncfm-core is typed (py.typed) and exposes a frontend-independent application layer. The simplest integration uses its factories, SQLite state, and built-in network adapters.

Add the dependency

Until a registry release is available, install from Git. Pin a tag or full commit for applications:

uv add "syncfm-core @ git+https://git.hirad.it/Hirad/syncfm-core.git@REVISION"

With pip:

python -m pip install "syncfm-core @ git+https://git.hirad.it/Hirad/syncfm-core.git@REVISION"

The package requires Python 3.12 or newer.

The package is licensed under GPL-3.0-or-later. If you distribute an application that incorporates it, review the GPL's source, license-notice, and corresponding-source requirements for your distribution model. This wiki is technical guidance, not legal advice.

Minimal direct integration

from pathlib import Path

from syncfm_core import (
    NetworkConfig,
    NetworkFactory,
    NetworkType,
    SyncRequest,
    create_sync_service,
)

factory = NetworkFactory()

source = factory.create(
    NetworkConfig(
        type=NetworkType.LASTFM,
        username="alice",
        password="source-password",
        api_key="source-api-key",
        api_secret="source-api-secret",
    )
)
target = factory.create(
    NetworkConfig(
        type=NetworkType.LIBREFM,
        username="alice",
        password="target-password",
        api_key="target-api-key",
        api_secret="target-api-secret",
    )
)

service = create_sync_service(Path("./state/syncfm.sqlite3"))
result = service.sync(
    source,
    target,
    SyncRequest(start_timestamp=1704067200),
)

print(result.session_id, result.synced_count)

Do not hard-code credentials in production. Use your application's secret manager or the profile service below.

Integrate through saved profiles

from pathlib import Path

from syncfm_core import (
    NetworkFactory,
    ProfileInput,
    SecretStorage,
    SyncRequest,
    create_profile_service,
    create_sync_service,
)
from syncfm_core.models import NetworkType

profiles = create_profile_service(
    Path("./config/profiles.toml"),
    keyring_service="my-product-syncfm",
)

profiles.save(
    ProfileInput(
        name="source",
        type=NetworkType.LASTFM,
        username="alice",
        api_key="public-key",
        api_secret="secret",
        session_key="existing-session-key",
        secret_storage=SecretStorage.KEYRING,
    )
)

# Create the "target" profile the same way, or provision it separately.

factory = NetworkFactory()
source = factory.create(profiles.resolve("source"))
target = factory.create(profiles.resolve("target"))

service = create_sync_service(Path("./state/syncfm.sqlite3"))
result = service.sync(
    source,
    target,
    SyncRequest(start_timestamp=1704067200),
    source_profile="source",
    target_profile="target",
)

Supplying both profile names is optional in the core API, but strongly recommended when profile-backed networks are used: it lets a later process discover which credentials are needed to resume. The pair is all-or-none.

Progress reporting

The callback runs synchronously in the thread performing the sync:

from syncfm_core import SyncProgress, create_sync_service

def report(progress: SyncProgress) -> None:
    current = progress.current_scrobble
    label = f"{current.artist} — {current.title}" if current else ""
    print(progress.phase, progress.completed, progress.total, label)

service = create_sync_service(
    "./syncfm.sqlite3",
    progress_callback=report,
)

Phase transitions report completed=0 and total=0. Fetching reports two units (source then target). Enrichment and submission report per-scrobble totals and may report a current scrobble before and after completion. Keep callbacks fast and exception-safe; callback exceptions are not swallowed.

Retry policy

Only NetworkConnectionError is retried. Authentication, invalid-response, and other network errors fail immediately.

from syncfm_core import RetryPolicy, create_sync_service

service = create_sync_service(
    "./syncfm.sqlite3",
    retry_policy=RetryPolicy(max_attempts=5, delay_seconds=2.0),
)

Retries apply to per-track metadata lookup and submission. Snapshot retrieval and recovery reconciliation are not wrapped in RetryService.

Resume and lifecycle management

pending = service.get_pending_sessions()

for session in pending:
    print(session.id, session.phase, session.source_profile, session.target_profile)

result = service.resume(
    pending[0].id,
    source,
    target,
    source_profile="source",
    target_profile="target",
)

# Or permanently remove local state:
service.discard(pending[0].id)

The supplied network name values and any stored profile names must match the original session. A successful sync deletes its session, snapshots, and queue through SQLite cascade rules after building the returned result.

Unlike the CLI, the core service does not enforce a single-pending-session policy. Coordinate access at the application level and use distinct databases when workloads should be isolated.

Implement a custom network

The engine uses structural protocols. A custom object need not subclass a SyncFM class; it must implement the MusicNetwork shape:

from syncfm_core import Scrobble, TrackMetadata

class MyNetwork:
    @property
    def name(self) -> str:
        return "MyNetwork [alice]"

    def get_scrobbles(
        self,
        start_timestamp: int | None = None,
        end_timestamp: int | None = None,
        limit: int | None = None,
    ) -> list[Scrobble]:
        # Return playback events, respecting inclusive bounds and limit.
        ...

    def get_latest_scrobble_timestamp(self) -> int | None:
        ...

    def get_track_metadata(self, artist: str, title: str) -> TrackMetadata:
        ...

    def submit_scrobble(self, scrobble: Scrobble) -> None:
        ...

Raise application-owned exceptions from syncfm_core.exceptions: use NetworkConnectionError for transient failures eligible for retry, AuthenticationError for rejected credentials, and InvalidNetworkResponseError for malformed responses.

The source is used as ScrobbleReader and MetadataProvider; the target is used as ScrobbleReader and ScrobbleWriter. Both are typed as the complete MusicNetwork protocol by the high-level service.

Advanced dependency injection

SyncService accepts repository and subservice dependencies in its constructor, and repository protocols are available from syncfm_core.protocols. This supports alternate persistence and isolated tests. The stable convenience path is create_sync_service; direct service construction couples the application to lower-level modules and should be pinned to a known package revision.

See Public API Reference and Synchronization Behavior for exact contracts.