Syncing

A pure-python client for the AnkiWeb sync protocol

AnkiWeb’s protocol is simpler than its lack of documentation suggests: JSON over POST, zstd-compressed, with auth in a header. A sync session walks a fixed sequence of endpoints under {endpoint}sync/: meta to compare collection state, then for a normal (delta) sync start, applyGraves, applyChanges, chunk (repeated), applyChunk (repeated), sanityCheck2 and finish. When the two sides have diverged too far, a full sync replaces one side’s whole sqlite file via upload or download. Everything here was built against Anki’s Rust implementation as the reference, and is tested below against a real Anki sync server, with Anki itself as the second client.

import tempfile
from fastcore.test import *
from google.protobuf.json_format import MessageToDict

The wire

Every request is a POST whose body is zstd-compressed JSON, with an anki-sync header carrying the protocol version (v, currently 11), the auth key (k, empty only for login), a client version string (c), and a random session id (s). Responses come back zstd-compressed too. A 308 means “use this other host from now on”: AnkiWeb spreads users across shards this way, so the new endpoint gets saved with the auth key.

---

source

SyncServer

def SyncServer(
    endpoint:NoneType=None, hkey:str=''
):

One AnkiWeb-protocol server (AnkiWeb itself, or self-hosted)

Schema 11 conversions

Here’s a wrinkle: although both sides store the modern schema, the delta protocol exchanges notetypes, decks and deck configs in the 2012 file format, JSON dicts nicknamed “schema 11”, for compatibility with older clients. So we need mappings between those dicts and our tables. Two rules matter for round-trip safety: unknown keys accumulate in an other JSON blob within the modern config so nothing is dropped, and a handful of fields changed units or sign along the way (autoplay becomes disable_autoplay, initial ease moves from thousands to a float, and so on).

---

source

nt_to_s11

def nt_to_s11(
    con, ntid
):

Convert notetype ntid’s rows to a schema-11 dict

---

source

nt_from_s11

def nt_from_s11(
    m
):

Convert a schema-11 notetype dict to (notetype_row, field_rows, template_rows)

---

source

deck_to_s11

def deck_to_s11(
    con, did
):

Convert deck did’s row to a schema-11 dict

---

source

deck_from_s11

def deck_from_s11(
    d
):

Convert a schema-11 deck dict to a decks table row (filtered decks unsupported: kept as-is if present locally)

---

source

dconf_to_s11

def dconf_to_s11(
    con, dcid
):

Convert deck config dcid’s row to a schema-11 dict

---

source

dconf_from_s11

def dconf_from_s11(
    d
):

Convert a schema-11 deck config dict to a deck_config table row

To test the conversions, we get genuine schema-11 dicts from Anki itself (its Python API still hands out models and decks in exactly this format), push them through *_from_s11, read them back with *_to_s11, and check every value Anki sent is preserved:

from anki.collection import Collection as AnkiCollection
import anki.lang

def sub_eq(a, b, path=''):
    "Assert every key/value in `a` appears in `b` (recursively, with numeric tolerance)"
    for k,v in a.items():
        w = b.get(k)
        if isinstance(v,dict): sub_eq(v, w or {}, f'{path}{k}.')
        elif isinstance(v,float) or isinstance(w,float): assert abs(v-(w or 0))<1e-6, f'{path}{k}: {v} vs {w}'
        elif isinstance(v,bool) or isinstance(w,bool): assert bool(v)==bool(w), f'{path}{k}: {v} vs {w}'
        else: assert v==w, f'{path}{k}: {v!r} vs {w!r}'

td = Path(tempfile.mkdtemp())
ac = AnkiCollection(str(td/'oracle.anki2'))
mcon = connect(td/'scratch.anki2')
mcon.execute(SCHEMA)
m11 = ac.models.by_name('Cloze')
row,flds,tmpls = nt_from_s11(m11)
mcon.execute('insert into notetypes values (?,?,?,?,?)', row)
mcon.executemany('insert into fields values (?,?,?,?)', flds)
mcon.executemany('insert into templates values (?,?,?,?,?,?)', tmpls)
back = nt_to_s11(mcon, row[0])
sub_eq(dict(m11), back)

d11 = ac.decks.get(1)
mcon.execute('insert into decks values (?,?,?,?,?,?)', deck_from_s11(d11))
sub_eq({k:v for k,v in d11.items() if k!='mid'}, deck_to_s11(mcon, 1))
c11 = ac.decks.get_config(1)
mcon.execute('insert into deck_config values (?,?,?,?,?)', dconf_from_s11(c11))
sub_eq(dict(c11), dconf_to_s11(mcon, 1))

Gathering and applying changes

A delta sync moves four kinds of data. Graves (deletion tombstones) go first in both directions. Then “unchunked changes”: notetypes, decks, deck configs and tags, sent whole. Then notes, cards and review log entries stream in chunks of positional JSON arrays. Rows pending sync are marked usn=-1; as each side sends them it rewrites the mark to the server’s current usn, so both ends converge on the same sequence number at finish.

The sync state machine

sync_collection ties it together, following the same sequence as Anki’s client. meta decides between nothing-to-do, a delta sync, and a full sync; a delta walks graves, changes, chunks down, chunks up, sanity check, finish. The sanity check compares object counts on both ends; on a mismatch we bump our schema time so the next sync is a full one, which is exactly what Anki does.

---

source

Collection.sync_collection

def sync_collection(
    srv
):

Delta-sync with srv, returning ‘noChanges’, ‘success’, or raising FullSyncRequired

---

source

Collection.sync_meta

def sync_meta():

Call self as a function.

---

source

SanityCheckFailed

def SanityCheckFailed(
    *args, **kwargs
):

Common base class for all non-exit exceptions.

---

source

FullSyncRequired

def FullSyncRequired(
    *args, **kwargs
):

Common base class for all non-exit exceptions.

---

source

SyncRequired

def SyncRequired(
    *args, **kwargs
):

Common base class for all non-exit exceptions.

Full sync and login

A full download returns the server’s whole sqlite file; upload sends ours (the server integrity-checks it before accepting, answering with the literal string OK). Auth is a host key from hostKey plus the endpoint any 308 redirect settled on, saved as JSON next to the collection.

---

source

Collection.sync

def sync(
    user:NoneType=None, passw:NoneType=None, endpoint:NoneType=None, upload:bool=False
):

Sync with AnkiWeb (or endpoint): logs in if needed, handles delta and full syncs

---

source

Collection.load_auth

def load_auth(
    endpoint:NoneType=None
):

A SyncServer from saved auth, or None

---

source

Collection.save_auth

def save_auth(
    srv
):

Call self as a function.

---

source

Collection.full_upload

def full_upload(
    srv
):

Replace the server’s collection with ours

---

source

Collection.full_download

def full_download(
    srv
):

Replace the local collection with the server’s copy

Against a real server

The anki wheel ships the real sync server (the same code AnkiWeb’s protocol front end is built from). That makes the test story write itself: fastanki syncs up to a genuine Anki server on localhost, then Anki, acting as a second client, syncs down and checks what arrived; then the same in reverse. The cells below tell that story but are marked eval: false: they need a server process, so tests/test_sync.py runs this exact sequence under pytest, where a fixture launches the server in a subprocess with a temp data folder and kills it afterwards. Nothing here talks to the real AnkiWeb.

Because “start a real server” is something the tests, these story cells, and any interactive session all need, it’s a library function: pick a free port, launch python -m anki.syncserver from the installed wheel, wait for it to answer, and register an atexit stop so a forgotten terminate() can’t leave orphans. It needs the anki package, which is a dev dependency only.

---

source

start_sync_server

def start_sync_server(
    base, # Folder for the server's data
    user:str='tester', # Username the server will accept
    passw:str='s3kret', # Password for `user`
):

Start the anki wheel’s sync server on a free port, returning (process, endpoint); it stops at exit, or sooner via .terminate()

server, EP = start_sync_server(td/'server')

Our client logs in, adds some content, and full-uploads (a fresh server and a fresh client have different schema stamps, so the first sync is always a full one):

col = Collection.open(td/'ours'/'collection.anki2')
n1 = col.add(Front='syncme', Back='please', tags='wire', deck='Sync::Deep')
ncz = col.add(model='Cloze', Text='{{c1::first}} {{c2::second}}')
test_fail(lambda: col.sync_collection(SyncServer(EP, hkey='bad')))
res = col.sync(user='tester', passw='s3kret', endpoint=EP, upload=True)
test_eq(res, 'full sync')

test_eq(col.sync(), 'noChanges')

Now Anki plays second client: log in with its own machinery, full-download, and inspect what we uploaded.

(td/'second').mkdir()
oc2 = AnkiCollection(str(td/'second'/'collection.anki2'))
auth = oc2.sync_login('tester', 's3kret', endpoint=EP)
st = oc2.sync_collection(auth, sync_media=False)
oc2.close_for_full_sync()
oc2.full_upload_or_download(auth=auth, server_usn=st.server_media_usn, upload=False)
oc2.reopen(after_full_sync=True)
test_eq(sorted(oc2.find_notes('')), sorted([n1.id, ncz.id]))
test_eq(oc2.find_notes('tag:wire'), [n1.id])
test_eq(oc2.find_notes('deck:Sync::Deep'), [n1.id])
onote = oc2.get_note(ncz.id)
test_eq(onote['Text'], '{{c1::first}} {{c2::second}}')

Then it makes changes of its own: a new note, a new deck, an edit and a deletion, delta-synced up with Anki’s client code:

theirs = oc2.new_note(oc2.models.by_name('Basic'))
theirs['Front'] = 'from the other side'
oc2.add_note(theirs, oc2.decks.add_normal_deck_with_name('Oracle').id)
on = oc2.get_note(n1.id)
on['Back'] = 'edited remotely'
oc2.update_note(on)
oc2.remove_notes([ncz.id])
st = oc2.sync_collection(auth, sync_media=False)
test_eq(st.required, st.NO_CHANGES)

And a delta sync brings all three changes to us, passing the server’s count check on the way:

test_eq(col.sync(), 'success')
test_eq(col.find_note_ids(Front='from the other side'), [theirs.id])
test_eq(col.get_note(n1.id)['Back'], 'edited remotely')
test_fail(lambda: col.get_note(ncz.id))
test_eq([d for d in col.decks() if d=='Oracle'], ['Oracle'])

Finally the reverse direction once more: our deletion and a new tagged note, verified by the oracle.

n2 = col.add(Front='round trip', tags=['back','forth'])
col.remove_notes(n1.id)
test_eq(col.sync(), 'success')
st = oc2.sync_collection(auth, sync_media=False)
oc2.close()
oc3 = AnkiCollection(str(td/'second'/'collection.anki2'))
test_eq(sorted(oc3.find_notes('')), sorted([theirs.id, n2.id]))
test_eq(oc3.find_notes('tag:forth'), [n2.id])
oc3.close()
col.close()
server.terminate()