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)
print("Source coverage verified:", result.verification.source_fully_synced)

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.

Post-sync verification uses a separate polling policy: the standard composition fetches fresh source history once and fresh target history up to three times, sleeping two seconds between incomplete checks. Network exceptions during these reads propagate immediately; this polling is intended to accommodate delayed visibility of accepted submissions, not transient request failures. Custom verification timing requires direct SyncService composition.

Verification results

Every successful sync() or resume() return includes fresh-history verification:

verification = result.verification

print(verification.start_timestamp, verification.end_timestamp)
print("Fresh source count:", verification.source_count)
print("Fresh target count:", verification.target_count)

for scrobble in verification.missing_on_target:
    print("Still missing:", scrobble.artist, scrobble.title, scrobble.timestamp)

source_fully_synced means every fresh source occurrence was found on the target. histories_match additionally requires that the source contain every target occurrence. Matching uses exact artist, title, and timestamp identities and respects duplicate occurrences.

An incomplete verification is returned normally rather than raising an exception. The core service does not retry another synchronization automatically; that interactive behavior belongs to the CLI.

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. After submission, fresh verification runs and the result is built before the session, snapshots, and queue are deleted through SQLite cascade rules. Local state is also deleted when verification reports missing target scrobbles, because an incomplete verification is still a completed operation 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, including a VerificationService, in its constructor, and repository protocols are available from syncfm_core.protocols. This supports alternate persistence, verification polling, 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.