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 zstandard, httpx, random, time, json
from fastcore.utils import *
from fastanki.schema import *
from fastanki.collection import *
from fastanki._proto import notetypes_pb2, decks_pb2, deck_config_pb2

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.

SYNC_VER = 11
CLIENT_VER = 'anki,26.05 (fastanki),python'

class SyncServer:
    "One AnkiWeb-protocol server (AnkiWeb itself, or self-hosted)"
    def __init__(self, endpoint=None, hkey=''):
        self.endpoint = endpoint or 'https://sync.ankiweb.net/'
        self.hkey,self.skey = hkey,f'{random.getrandbits(32):08x}'
        self.client = httpx.Client(timeout=60)
    def __call__(self, method, obj=None, data=None):
        "POST `obj` (or raw `data`) to sync/`method`, returning the decompressed response body"
        body = zstandard.compress(data if data is not None else json.dumps(obj or {}).encode())
        hdr = json.dumps(dict(v=SYNC_VER, k=self.hkey, c=CLIENT_VER, s=self.skey))
        r = self.client.post(f'{self.endpoint}sync/{method}', content=body, headers={'anki-sync': hdr, 'content-type': 'application/octet-stream'})
        if r.status_code==308:
            self.endpoint = r.headers['location'].rstrip('/') + '/'
            return self(method, obj, data)
        r.raise_for_status()
        return zstandard.ZstdDecompressor().decompressobj().decompress(r.content) if 'anki-original-size' in r.headers else r.content
    def json(self, method, obj=None): return json.loads(self(method, obj))
    def login(self, user, passw):
        "Exchange credentials for a host key, stored on this server object"
        self.hkey = self.json('hostKey', dict(u=user, p=passw))['key']
        return self.hkey

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).

def _other_in(d, reserved):
    "JSON-encode unreserved keys of `d`, Anki's mechanism for preserving unknown fields"
    o = {k:v for k,v in d.items() if k not in reserved}
    return json.dumps(o).encode() if o else b''

def _other_out(b, reserved):
    if not b: return {}
    return {k:v for k,v in json.loads(b).items() if k not in reserved}

_NT_KEYS = {'id','name','type','mod','usn','sortf','did','tmpls','flds','css','latexPre','latexPost','latexsvg','req','originalStockKind','originalId'}
_FLD_KEYS = {'name','ord','sticky','rtl','plainText','font','size','collapsed','description','excludeFromSearch','id','tag','preventDeletion'}
_TMPL_KEYS = {'name','ord','did','afmt','bafmt','qfmt','bqfmt','bfont','bsize','id'}
_REQKIND = {'none':0,'any':1,'all':2}

def nt_from_s11(m):
    "Convert a schema-11 notetype dict to (notetype_row, field_rows, template_rows)"
    c = notetypes_pb2.Notetype.Config(kind=int(m['type']), sort_field_idx=int(m['sortf']), css=m.get('css',''),
        target_deck_id_unused=int(m.get('did') or 0), latex_pre=m.get('latexPre',''), latex_post=m.get('latexPost',''),
        latex_svg=bool(m.get('latexsvg')), original_stock_kind=int(m.get('originalStockKind',0)),
        other=_other_in(m, _NT_KEYS))
    if m.get('originalId') is not None: c.original_id = int(m['originalId'])
    for r in m.get('req') or []:
        q = c.reqs.add(); q.card_ord,q.kind = int(r[0]),_REQKIND[r[1]]; q.field_ords.extend(map(int,r[2]))
    ntid = int(m['id'])
    flds = []
    for i,f in enumerate(sorted(m['flds'], key=lambda f:f.get('ord') or 0)):
        fc = notetypes_pb2.Notetype.Field.Config(sticky=bool(f.get('sticky')), rtl=bool(f.get('rtl')),
            plain_text=bool(f.get('plainText')), font_name=f.get('font','Arial'), font_size=int(f.get('size',20)),
            description=f.get('description',''), collapsed=bool(f.get('collapsed')),
            exclude_from_search=bool(f.get('excludeFromSearch')), other=_other_in(f, _FLD_KEYS))
        if f.get('id') is not None: fc.id = int(f['id'])
        if f.get('tag') is not None: fc.tag = int(f['tag'])
        if f.get('preventDeletion'): fc.prevent_deletion = True
        flds.append((ntid, i, f['name'], fc.SerializeToString()))
    tmpls = []
    for i,t in enumerate(sorted(m['tmpls'], key=lambda t:t.get('ord') or 0)):
        tc = notetypes_pb2.Notetype.Template.Config(q_format=t.get('qfmt',''), a_format=t.get('afmt',''),
            q_format_browser=t.get('bqfmt',''), a_format_browser=t.get('bafmt',''),
            target_deck_id=int(t.get('did') or 0), browser_font_name=t.get('bfont',''),
            browser_font_size=int(t.get('bsize') or 0), other=_other_in(t, _TMPL_KEYS))
        if t.get('id') is not None: tc.id = int(t['id'])
        tmpls.append((ntid, i, t['name'], 0, 0, tc.SerializeToString()))
    row = (ntid, m['name'], int(m['mod']), int(m['usn']), c.SerializeToString())
    return row, flds, tmpls

def nt_to_s11(con, ntid):
    "Convert notetype `ntid`'s rows to a schema-11 dict"
    name,mtime,usn,blob = con.execute('select name, mtime_secs, usn, config from notetypes where id=?',(ntid,)).fetchone()
    c = notetypes_pb2.Notetype.Config(); c.ParseFromString(blob)
    m = dict(id=ntid, name=name, type=c.kind, mod=mtime, usn=usn, sortf=c.sort_field_idx,
        did=c.target_deck_id_unused or None, css=c.css, latexPre=c.latex_pre, latexPost=c.latex_post,
        latexsvg=c.latex_svg, originalStockKind=c.original_stock_kind,
        req=[[r.card_ord, ['none','any','all'][r.kind], list(r.field_ords)] for r in c.reqs],
        flds=[], tmpls=[], **_other_out(c.other, _NT_KEYS))
    if c.HasField('original_id'): m['originalId'] = c.original_id
    for ord_,fname,blob in con.execute('select ord, name, config from fields where ntid=? order by ord',(ntid,)):
        fc = notetypes_pb2.Notetype.Field.Config(); fc.ParseFromString(blob)
        f = dict(name=fname, ord=ord_, sticky=fc.sticky, rtl=fc.rtl, plainText=fc.plain_text,
            font=fc.font_name, size=fc.font_size, description=fc.description, collapsed=fc.collapsed,
            excludeFromSearch=fc.exclude_from_search, **_other_out(fc.other, _FLD_KEYS))
        if fc.HasField('id'): f['id'] = fc.id
        if fc.HasField('tag'): f['tag'] = fc.tag
        f['preventDeletion'] = fc.prevent_deletion
        m['flds'].append(f)
    for ord_,tname,blob in con.execute('select ord, name, config from templates where ntid=? order by ord',(ntid,)):
        tc = notetypes_pb2.Notetype.Template.Config(); tc.ParseFromString(blob)
        t = dict(name=tname, ord=ord_, qfmt=tc.q_format, afmt=tc.a_format, bqfmt=tc.q_format_browser,
            bafmt=tc.a_format_browser, did=tc.target_deck_id or None, bfont=tc.browser_font_name,
            bsize=tc.browser_font_size, **_other_out(tc.other, _TMPL_KEYS))
        if tc.HasField('id'): t['id'] = tc.id
        m['tmpls'].append(t)
    return m

_DECK_KEYS = {'id','mod','name','usn','lrnToday','revToday','newToday','timeToday','collapsed','browserCollapsed',
              'desc','md','dyn','conf','extendNew','extendRev','reviewLimit','newLimit','reviewLimitToday','newLimitToday','desiredRetention'}

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)"
    assert not int(d.get('dyn') or 0), "Filtered decks not supported"
    today = {k: d.get(k) or [0,0] for k in ('lrnToday','revToday','newToday','timeToday')}
    max_day = max(today['timeToday'][0], today['newToday'][0], today['revToday'][0])
    amt = lambda k: today[k][1] if today[k][0]==max_day else 0
    common = decks_pb2.Deck.Common(study_collapsed=bool(d.get('collapsed')), browser_collapsed=bool(d.get('browserCollapsed')),
        last_day_studied=max(max_day,0), new_studied=amt('newToday'), review_studied=amt('revToday'),
        learning_studied=amt('lrnToday'), milliseconds_studied=today['timeToday'][1],
        other=_other_in(d, _DECK_KEYS))
    normal = decks_pb2.Deck.Normal(config_id=int(d.get('conf',1)), extend_new=max(int(d.get('extendNew') or 0),0),
        extend_review=max(int(d.get('extendRev') or 0),0), markdown_description=bool(d.get('md')),
        description=d.get('desc',''))
    for k,f in [('reviewLimit','review_limit'),('newLimit','new_limit')]:
        if d.get(k) is not None: setattr(normal, f, int(d[k]))
    if d.get('desiredRetention') is not None: normal.desired_retention = int(d['desiredRetention'])/100.0
    kind = decks_pb2.Deck.KindContainer(normal=normal)
    return (int(d['id']), d['name'].replace('::','\x1f'), int(d['mod']), int(d['usn']), common.SerializeToString(), kind.SerializeToString())

def deck_to_s11(con, did):
    "Convert deck `did`'s row to a schema-11 dict"
    did,name,mtime,usn,cb,kb = con.execute('select * from decks where id=?',(did,)).fetchone()
    c = decks_pb2.Deck.Common(); c.ParseFromString(cb)
    k = decks_pb2.Deck.KindContainer(); k.ParseFromString(kb)
    assert k.WhichOneof('kind')=='normal', "Filtered decks not supported"
    n = k.normal
    day = c.last_day_studied
    d = dict(id=did, mod=mtime, name=name.replace('\x1f','::'), usn=usn,
        lrnToday=[day,c.learning_studied], revToday=[day,c.review_studied],
        newToday=[day,c.new_studied], timeToday=[day,c.milliseconds_studied],
        collapsed=c.study_collapsed, browserCollapsed=c.browser_collapsed,
        desc=n.description, dyn=0, conf=n.config_id or 1,
        extendNew=n.extend_new, extendRev=n.extend_review, **_other_out(c.other, _DECK_KEYS))
    if n.markdown_description: d['md'] = True
    if n.HasField('review_limit'): d['reviewLimit'] = n.review_limit
    if n.HasField('new_limit'): d['newLimit'] = n.new_limit
    if n.HasField('desired_retention'): d['desiredRetention'] = round(n.desired_retention*100)
    return d

_DCONF_KEYS = {'id','mod','name','usn','maxTaken','autoplay','timer','replayq','new','rev','lapse','dyn',
    'newMix','newPerDayMinimum','interdayLearningMix','reviewOrder','newSortOrder','newGatherPriority',
    'buryInterdayLearning','fsrsWeights','fsrsParams5','fsrsParams6','desiredRetention','ignoreRevlogsBeforeDate',
    'easyDaysPercentages','stopTimerOnAnswer','secondsToShowQuestion','secondsToShowAnswer','questionAction',
    'answerAction','waitForAudio','sm2Retention','weightSearch'}
_NEW_KEYS = {'bury','delays','initialFactor','ints','order','perDay'}
_REV_KEYS = {'bury','ease4','ivlFct','maxIvl','perDay','hardFactor'}
_LAPSE_KEYS = {'delays','leechAction','leechFails','minInt','mult'}

def dconf_from_s11(d):
    "Convert a schema-11 deck config dict to a `deck_config` table row"
    new,rev,lapse = (d.get(k) or {} for k in ('new','rev','lapse'))
    other = {k:v for k,v in d.items() if k not in _DCONF_KEYS}
    for k,sub,keys in [('new',new,_NEW_KEYS),('rev',rev,_REV_KEYS),('lapse',lapse,_LAPSE_KEYS)]:
        o = {kk:vv for kk,vv in sub.items() if kk not in keys}
        if o: other[k] = o
    ints = (new.get('ints') or [1,4])
    c = deck_config_pb2.DeckConfig.Config(
        learn_steps=[float(x) for x in new.get('delays') or [1.0,10.0]],
        relearn_steps=[float(x) for x in lapse.get('delays') or [10.0]],
        new_per_day=int(new.get('perDay',20)), reviews_per_day=int(rev.get('perDay',200)),
        new_per_day_minimum=int(d.get('newPerDayMinimum') or 0),
        initial_ease=int(new.get('initialFactor',2500))/1000.0,
        easy_multiplier=float(rev.get('ease4',1.3)), hard_multiplier=float(rev.get('hardFactor',1.2)),
        lapse_multiplier=float(lapse.get('mult',0.0)), interval_multiplier=float(rev.get('ivlFct',1.0)),
        maximum_review_interval=int(rev.get('maxIvl',36500)), minimum_lapse_interval=int(lapse.get('minInt',1)),
        graduating_interval_good=int(ints[0]), graduating_interval_easy=int(ints[1]),
        new_card_insert_order=int(new.get('order',1)),
        new_card_gather_priority=int(d.get('newGatherPriority') or 0),
        new_card_sort_order=int(d.get('newSortOrder') or 0),
        review_order=int(d.get('reviewOrder') or 0), new_mix=int(d.get('newMix') or 0),
        interday_learning_mix=int(d.get('interdayLearningMix') or 0),
        leech_action=int(lapse.get('leechAction',1)), leech_threshold=int(lapse.get('leechFails',8)),
        disable_autoplay=not d.get('autoplay',True), cap_answer_time_to_secs=max(int(d.get('maxTaken',60)),0),
        show_timer=bool(d.get('timer')), stop_timer_on_answer=bool(d.get('stopTimerOnAnswer')),
        seconds_to_show_question=float(d.get('secondsToShowQuestion') or 0),
        seconds_to_show_answer=float(d.get('secondsToShowAnswer') or 0),
        question_action=int(d.get('questionAction') or 0), answer_action=int(d.get('answerAction') or 0),
        wait_for_audio=d.get('waitForAudio',True), skip_question_when_replaying_answer=not d.get('replayq',True),
        bury_new=bool(new.get('bury')), bury_reviews=bool(rev.get('bury')),
        bury_interday_learning=bool(d.get('buryInterdayLearning')),
        fsrs_params_4=[float(x) for x in d.get('fsrsWeights') or []],
        fsrs_params_5=[float(x) for x in d.get('fsrsParams5') or []],
        fsrs_params_6=[float(x) for x in d.get('fsrsParams6') or []],
        ignore_revlogs_before_date=d.get('ignoreRevlogsBeforeDate',''),
        easy_days_percentages=[float(x) for x in d.get('easyDaysPercentages') or [1.0]*7],
        desired_retention=float(d.get('desiredRetention') or 0.9),
        historical_retention=float(d.get('sm2Retention') or 0.9),
        param_search=d.get('weightSearch',''),
        other=json.dumps(other).encode() if other else b'')
    return (int(d['id']), d['name'], int(d['mod']), int(d['usn']), c.SerializeToString())

def dconf_to_s11(con, dcid):
    "Convert deck config `dcid`'s row to a schema-11 dict"
    dcid,name,mtime,usn,blob = con.execute('select * from deck_config where id=?',(dcid,)).fetchone()
    c = deck_config_pb2.DeckConfig.Config(); c.ParseFromString(blob)
    other = json.loads(c.other) if c.other else {}
    d = dict(id=dcid, mod=mtime, name=name, usn=usn, maxTaken=c.cap_answer_time_to_secs,
        autoplay=not c.disable_autoplay, timer=1 if c.show_timer else 0, replayq=not c.skip_question_when_replaying_answer,
        new=dict(bury=c.bury_new, delays=list(c.learn_steps), initialFactor=round(c.initial_ease*1000),
            ints=[c.graduating_interval_good, c.graduating_interval_easy, 0], order=c.new_card_insert_order,
            perDay=c.new_per_day, **(other.pop('new',{}) if isinstance(other.get('new'),dict) else {})),
        rev=dict(bury=c.bury_reviews, ease4=c.easy_multiplier, ivlFct=c.interval_multiplier,
            maxIvl=c.maximum_review_interval, perDay=c.reviews_per_day, hardFactor=c.hard_multiplier,
            **(other.pop('rev',{}) if isinstance(other.get('rev'),dict) else {})),
        lapse=dict(delays=list(c.relearn_steps), leechAction=c.leech_action, leechFails=c.leech_threshold,
            minInt=c.minimum_lapse_interval, mult=c.lapse_multiplier,
            **(other.pop('lapse',{}) if isinstance(other.get('lapse'),dict) else {})),
        dyn=False, newMix=c.new_mix, newPerDayMinimum=c.new_per_day_minimum,
        interdayLearningMix=c.interday_learning_mix, reviewOrder=c.review_order,
        newSortOrder=c.new_card_sort_order, newGatherPriority=c.new_card_gather_priority,
        buryInterdayLearning=c.bury_interday_learning, fsrsWeights=list(c.fsrs_params_4),
        fsrsParams5=list(c.fsrs_params_5), fsrsParams6=list(c.fsrs_params_6),
        desiredRetention=c.desired_retention, ignoreRevlogsBeforeDate=c.ignore_revlogs_before_date,
        easyDaysPercentages=list(c.easy_days_percentages), stopTimerOnAnswer=c.stop_timer_on_answer,
        secondsToShowQuestion=c.seconds_to_show_question, secondsToShowAnswer=c.seconds_to_show_answer,
        questionAction=c.question_action, answerAction=c.answer_action, waitForAudio=c.wait_for_audio,
        sm2Retention=c.historical_retention, weightSearch=c.param_search)
    d.update(other)
    return d

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.

GRAVE_CARD,GRAVE_NOTE,GRAVE_DECK = 0,1,2

@patch
def _pending_graves(self:Collection, new_usn):
    "Pending graves as {cards,notes,decks}, rewriting their usn to `new_usn`"
    g = dict(cards=[], notes=[], decks=[])
    for oid,typ in self.q('select oid, type from graves where usn=-1'):
        g[['cards','notes','decks'][typ]].append(oid)
    self.con.execute('update graves set usn=? where usn=-1', (new_usn,))
    return g

def _chunk_graves(g):
    "Yield grave sub-dicts of at most CHUNK_SIZE oids (cards, then notes, then decks), like Anki's take_chunk"
    items = [(k,o) for k in ('cards','notes','decks') for o in g[k]]
    for i in range(0, len(items), CHUNK_SIZE):
        c = dict(cards=[], notes=[], decks=[])
        for k,o in items[i:i+CHUNK_SIZE]: c[k].append(o)
        yield c

@patch
def _apply_graves(self:Collection, g, new_usn):
    for cid in g.get('cards') or []:
        self.con.execute('delete from cards where id=?',(cid,))
        self.con.execute('insert or replace into graves values (?,0,?)',(cid,new_usn))
    for nid in g.get('notes') or []:
        self.con.execute('delete from notes where id=?',(nid,))
        self.con.execute('insert or replace into graves values (?,1,?)',(nid,new_usn))
    for did in g.get('decks') or []:
        self.con.execute('delete from decks where id=?',(did,))
        self.con.execute('insert or replace into graves values (?,2,?)',(did,new_usn))

@patch
def _local_changes(self:Collection, new_usn, local_newer):
    "Pending unchunked changes in wire format, rewriting local usns to `new_usn`"
    models = [nt_to_s11(self.con, r[0]) for r in self.q('select id from notetypes where usn=-1')]
    decks = [deck_to_s11(self.con, r[0]) for r in self.q('select id from decks where usn=-1')]
    dconf = [dconf_to_s11(self.con, r[0]) for r in self.q('select id from deck_config where usn=-1')]
    tags = [r[0] for r in self.q('select tag from tags where usn=-1')]
    for m in models+decks+dconf: m['usn'] = new_usn
    for t in ['notetypes','decks','deck_config','tags']:
        self.con.execute(f'update {t} set usn=? where usn=-1', (new_usn,))
    changes = dict(models=models, decks=[decks,dconf], tags=tags)
    if local_newer:
        changes['conf'] = {k: json.loads(v) for k,v in self.q('select key, val from config')}
        changes['crt'] = self.q1('select crt from col')
    return changes

@patch
def _apply_changes(self:Collection, ch, new_usn):
    "Apply the server's unchunked changes, skipping any object we hold a newer copy of (Anki's mtime merge)"
    def stale(table, oid, mtime):
        cur = self.q1(f'select mtime_secs from {table} where id=?', oid)
        return cur is not None and cur > mtime
    for m in ch.get('models') or []:
        row,flds,tmpls = nt_from_s11(m)
        if stale('notetypes', row[0], row[2]): continue
        self.con.execute('delete from fields where ntid=?',(row[0],))
        self.con.execute('delete from templates where ntid=?',(row[0],))
        self.con.execute('insert or replace into notetypes values (?,?,?,?,?)', row)
        self.con.executemany('insert into fields values (?,?,?,?)', flds)
        self.con.executemany('insert into templates values (?,?,?,?,?,?)', tmpls)
    decks,dconf = ch.get('decks') or [[],[]]
    for d in decks:
        r = deck_from_s11(d)
        if not stale('decks', r[0], r[2]): self.con.execute('insert or replace into decks values (?,?,?,?,?,?)', r)
    for c in dconf:
        r = dconf_from_s11(c)
        if not stale('deck_config', r[0], r[2]): self.con.execute('insert or replace into deck_config values (?,?,?,?,?)', r)
    for t in ch.get('tags') or []:
        self.con.execute('insert into tags values (?,?,0,NULL) on conflict(tag) do update set usn=excluded.usn', (t, new_usn))
    if ch.get('conf') is not None:
        self.con.execute('delete from config')
        for k,v in ch['conf'].items(): self.con.execute('insert into config values (?,0,0,?)',(k, json.dumps(v).encode()))
    if ch.get('crt') is not None: self.con.execute('update col set crt=?',(ch['crt'],))

CHUNK_SIZE = 250
_NOTE_COLS = 'id, guid, mid, mod, usn, tags, flds, sfld, csum, flags, data'
_CARD_COLS = 'id, nid, did, ord, mod, usn, type, queue, due, ivl, factor, reps, lapses, left, odue, odid, flags, data'
_REVLOG_COLS = 'id, cid, usn, ease, ivl, lastIvl, factor, time, type'

@patch
def _register_note_tags(self:Collection):
    "Ensure every tag on a pending note exists in the tags table, so new tags propagate"
    for (tags,) in self.q('select tags from notes where usn=-1'):
        for t in tags.split():
            self.con.execute('insert or ignore into tags values (?,-1,0,NULL)',(t,))

@patch
def _chunkable_ids(self:Collection):
    return {t: [r[0] for r in self.q(f'select id from {t} where usn=-1')] for t in ('revlog','cards','notes')}

@patch
def _take_chunk(self:Collection, ids, new_usn):
    "Build one wire chunk (positional row arrays) from pending `ids`, rewriting their usn"
    chunk,n = {},0
    for t,cols in [('revlog',_REVLOG_COLS),('cards',_CARD_COLS),('notes',_NOTE_COLS)]:
        take,ids[t] = ids[t][:CHUNK_SIZE-n],ids[t][CHUNK_SIZE-n:]
        n += len(take)
        rows = []
        for oid in take:
            r = list(self.con.execute(f'select {cols} from {t} where id=?',(oid,)).fetchone())
            r[cols.split(', ').index('usn')] = new_usn
            if t=='notes': r[7],r[8] = '',''  # sfld/csum are recomputed by the receiver
            rows.append(r)
        if rows: chunk[t] = rows
        self.con.executemany(f'update {t} set usn=? where id=?', [(new_usn,i) for i in take])
    chunk['done'] = not any(ids.values())
    return chunk

@patch
def _apply_chunk(self:Collection, chunk):
    "Apply a server chunk: newer rows win over local pending ones"
    for r in chunk.get('revlog') or []:
        self.con.execute('insert or ignore into revlog values (?,?,?,?,?,?,?,?,?)', r)
    for r in chunk.get('cards') or []:
        cur = self.con.execute('select usn, mod from cards where id=?',(r[0],)).fetchone()
        if cur and cur[0]==-1 and cur[1]>=r[4]: continue
        self.con.execute('insert or replace into cards values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)', r)
    for r in chunk.get('notes') or []:
        cur = self.con.execute('select usn, mod from notes where id=?',(r[0],)).fetchone()
        if cur and cur[0]==-1 and cur[1]>=r[3]: continue
        flds = r[6]
        vals = flds.split('\x1f')
        sortf = 0
        blob = self.con.execute('select config from notetypes where id=?',(r[2],)).fetchone()
        if blob:
            cfg = notetypes_pb2.Notetype.Config(); cfg.ParseFromString(blob[0])
            sortf = min(cfg.sort_field_idx, len(vals)-1)
        r = list(r)
        r[7],r[8] = strip_html_media(vals[sortf]), field_csum(vals[0])
        self.con.execute('insert or replace into notes values (?,?,?,?,?,?,?,?,?,?,?)', r)

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.

class SyncRequired(Exception): pass
class FullSyncRequired(SyncRequired): pass
class SanityCheckFailed(Exception): pass

@patch
def _sanity_counts(self:Collection):
    c = lambda t: self.q1(f'select count() from {t}')
    return [[0,0,0], c('cards'), c('notes'), c('revlog'), c('graves'), c('notetypes'), c('decks'), c('deck_config')]

@patch
def sync_meta(self:Collection):
    mod,scm,usn,crt = self.q('select mod, scm, usn, crt from col')[0]
    empty = not self.q1('select exists(select 1 from cards)')
    return dict(mod=mod, scm=scm, usn=usn, ts=int(time.time()), msg='', cont=True, hostNum=0, empty=empty)

@patch
def sync_collection(self:Collection, srv):
    "Delta-sync with `srv`, returning 'noChanges', 'success', or raising `FullSyncRequired`"
    local = self.sync_meta()
    remote = srv.json('meta', dict(v=SYNC_VER, cv=CLIENT_VER))
    if remote.get('msg'): print(remote['msg'])
    if not remote.get('cont', True): raise Exception(f"Server refused sync: {remote.get('msg')}")
    if remote['mod']==local['mod']: return 'noChanges'
    if remote['scm']!=local['scm']: raise FullSyncRequired(local, remote)
    local_newer = local['mod']>remote['mod']
    server_usn = remote['usn']
    try:
        self.con.execute('BEGIN IMMEDIATE')
        rg = srv.json('start', dict(minUsn=local['usn'], lnewer=local_newer))
        for gchunk in _chunk_graves(self._pending_graves(server_usn)):
            srv.json('applyGraves', dict(chunk=gchunk))
        self._apply_graves(rg, server_usn)
        self._register_note_tags()
        rch = srv.json('applyChanges', dict(changes=self._local_changes(server_usn, local_newer)))
        self._apply_changes(rch, server_usn)
        while True:
            chunk = srv.json('chunk')
            self._apply_chunk(chunk)
            if chunk.get('done'): break
        ids = self._chunkable_ids()
        while True:
            chunk = self._take_chunk(ids, server_usn)
            srv.json('applyChunk', dict(chunk=chunk))
            if chunk['done']: break
        resp = srv.json('sanityCheck2', dict(client=self._sanity_counts()))
        if resp['status']!='ok': raise SanityCheckFailed(str(resp))
        new_mtime = srv.json('finish')
        self.con.execute('update col set mod=?, ls=?, usn=?',(new_mtime, new_mtime, server_usn+1))
        self.con.execute('COMMIT')
        return 'success'
    except Exception as e:
        self.con.execute('ROLLBACK')
        try: srv.json('abort')
        except Exception: pass
        # a sanity mismatch means our object counts diverged from the server's; bump scm to force a full sync next time (Anki's recovery path)
        if isinstance(e, SanityCheckFailed):
            self.con.execute('update col set scm=?',(now_ms(),))
        raise

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.

@patch
def _clear_wal(self:Collection):
    "Checkpoint and remove WAL sidecar files, so `self.path` alone is the whole database"
    self.con.execute('pragma wal_checkpoint(truncate)')

@patch
def full_download(self:Collection, srv):
    "Replace the local collection with the server's copy"
    data = srv('download')
    self._clear_wal()
    self.con.close()
    tmp = self.path.with_suffix('.tmp')
    tmp.write_bytes(data)
    con = connect(tmp)
    assert con.execute('select ver from col').fetchone(), "downloaded file failed to open"
    # keep the downloaded file's usn (the server's); zeroing it would make the next delta needlessly re-fetch the whole collection
    con.execute('update col set ls=mod')
    con.close()
    for ext in ['-wal','-shm']: Path(str(tmp)+ext).unlink(missing_ok=True)
    tmp.replace(self.path)
    for ext in ['-wal','-shm']: Path(str(self.path)+ext).unlink(missing_ok=True)
    self.con = connect(self.path)

@patch
def full_upload(self:Collection, srv):
    "Replace the server's collection with ours"
    self.con.execute('update col set scm=?, mod=?, ls=?, usn=usn+1',(now_ms(), now_ms(), now_ms()))
    # Anki's before_upload: notes/cards/revlog keep their usns (only pending -1 cleared), but decks/notetypes/deck_config/tags all reset to 0; col usn increments
    for t in ['notes','cards','revlog']:
        self.con.execute(f'update {t} set usn=0 where usn=-1')
    for t in ['notetypes','decks','deck_config','tags']:
        self.con.execute(f'update {t} set usn=0 where usn!=0')
    self.con.execute('delete from graves')
    self._clear_wal()
    resp = srv('upload', data=self.path.read_bytes())
    if resp!=b'OK': raise Exception(f"Upload failed: {resp[:200]}")

@patch(as_prop=True)
def _auth_path(self:Collection): return self.path.with_name('auth.json')

@patch
def save_auth(self:Collection, srv): self._auth_path.write_text(json.dumps(dict(hkey=srv.hkey, endpoint=srv.endpoint)))

@patch
def load_auth(self:Collection, endpoint=None):
    "A `SyncServer` from saved auth, or None"
    if not self._auth_path.exists(): return None
    a = json.loads(self._auth_path.read_text())
    return SyncServer(endpoint or a['endpoint'], a['hkey'])

@patch
def sync(self:Collection, user=None, passw=None, endpoint=None, upload=False):
    "Sync with AnkiWeb (or `endpoint`): logs in if needed, handles delta and full syncs"
    srv = self.load_auth(endpoint)
    if srv is None or (user and not srv.hkey):
        if not (user and passw): raise ValueError("No saved auth; pass user and passw")
        srv = SyncServer(endpoint)
        srv.login(user, passw)
    try: res = self.sync_collection(srv)
    except FullSyncRequired as e:
        local,remote = e.args
        if upload:
            if local['empty'] and not remote['empty']:
                raise ValueError("Refusing to replace a non-empty server collection with an empty local one") from None
            self.full_upload(srv)
        else:
            if remote['empty'] and not local['empty']:
                raise ValueError("Server collection is empty but local isn't; pass upload=True to replace the server copy") from None
            self.full_download(srv)
        res = 'full sync'
    self.save_auth(srv)
    return res

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.

import subprocess, sys

PORT = 27183
env = dict(SYNC_BASE=str(td/'server'), SYNC_USER1='tester:s3kret', SYNC_HOST='127.0.0.1', SYNC_PORT=str(PORT))
server = subprocess.Popen([sys.executable,'-m','anki.syncserver'], env=os.environ|env)
EP = f'http://127.0.0.1:{PORT}/'
for _ in range(50):
    try: httpx.get(EP); break
    except httpx.ConnectError: time.sleep(0.1)

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()