Collection files

Reading and writing Anki’s schema-18 sqlite collections, no Anki required

An .anki2 collection is a single sqlite file. Anki’s desktop app drives it through a Rust library, but nothing about the file needs Rust: it’s ordinary tables of notes, cards and review history, plus a handful of columns holding small protobuf messages (notetype, deck and deck-config settings). This module creates and connects to those files directly. The protobuf classes in fastanki._proto are generated from Anki’s own .proto files, so the blobs we write are byte-compatible with what Anki writes.

import apsw, time, random, json
from apsw import bestpractice
from fastcore.utils import *
from fastanki._proto import notetypes_pb2, decks_pb2, deck_config_pb2
STOCK = notetypes_pb2.StockNotetype.OriginalStockKind

from fastcore.test import *
import tempfile
from anki.collection import Collection as AnkiCollection
from google.protobuf.json_format import MessageToDict

Throughout this notebook we check our work against Anki itself: the anki package is a development-time dependency only, used as an oracle. If Anki can open a file we created, add a note to it, and pass its own database check, we got the details right.

The schema

The DDL below is dumped straight from a collection created by Anki 26.05 (the test that follows re-dumps and compares, so drift shows up here rather than in the field). Two details matter. The col table is a leftover from the old single-row JSON format: schema 18 keeps the row but stores empty strings in the JSON columns, with the real data moved to proper tables. And several columns are declared COLLATE unicase, a custom case-insensitive collation that Anki’s Rust registers with sqlite; we must register our own equivalent before any query that compares those columns.

SCHEMA = r"""
CREATE TABLE cards (
  id integer PRIMARY KEY,
  nid integer NOT NULL,
  did integer NOT NULL,
  ord integer NOT NULL,
  mod integer NOT NULL,
  usn integer NOT NULL,
  type integer NOT NULL,
  queue integer NOT NULL,
  due integer NOT NULL,
  ivl integer NOT NULL,
  factor integer NOT NULL,
  reps integer NOT NULL,
  lapses integer NOT NULL,
  left integer NOT NULL,
  odue integer NOT NULL,
  odid integer NOT NULL,
  flags integer NOT NULL,
  data text NOT NULL
);
CREATE TABLE col (
  id integer PRIMARY KEY,
  crt integer NOT NULL,
  mod integer NOT NULL,
  scm integer NOT NULL,
  ver integer NOT NULL,
  dty integer NOT NULL,
  usn integer NOT NULL,
  ls integer NOT NULL,
  conf text NOT NULL,
  models text NOT NULL,
  decks text NOT NULL,
  dconf text NOT NULL,
  tags text NOT NULL
);
CREATE TABLE config (
  KEY text NOT NULL PRIMARY KEY,
  usn integer NOT NULL,
  mtime_secs integer NOT NULL,
  val blob NOT NULL
) without rowid;
CREATE TABLE deck_config (
  id integer PRIMARY KEY NOT NULL,
  name text NOT NULL COLLATE unicase,
  mtime_secs integer NOT NULL,
  usn integer NOT NULL,
  config blob NOT NULL
);
CREATE TABLE decks (
  id integer PRIMARY KEY NOT NULL,
  name text NOT NULL COLLATE unicase,
  mtime_secs integer NOT NULL,
  usn integer NOT NULL,
  common blob NOT NULL,
  kind blob NOT NULL
);
CREATE TABLE fields (
  ntid integer NOT NULL,
  ord integer NOT NULL,
  name text NOT NULL COLLATE unicase,
  config blob NOT NULL,
  PRIMARY KEY (ntid, ord)
) without rowid;
CREATE TABLE graves (
  oid integer NOT NULL,
  type integer NOT NULL,
  usn integer NOT NULL,
  PRIMARY KEY (oid, type)
) WITHOUT ROWID;
CREATE TABLE notes (
  id integer PRIMARY KEY,
  guid text NOT NULL,
  mid integer NOT NULL,
  mod integer NOT NULL,
  usn integer NOT NULL,
  tags text NOT NULL,
  flds text NOT NULL,
  -- The use of type integer for sfld is deliberate, because it means that integer values in this
  -- field will sort numerically.
  sfld integer NOT NULL,
  csum integer NOT NULL,
  flags integer NOT NULL,
  data text NOT NULL
);
CREATE TABLE notetypes (
  id integer NOT NULL PRIMARY KEY,
  name text NOT NULL COLLATE unicase,
  mtime_secs integer NOT NULL,
  usn integer NOT NULL,
  config blob NOT NULL
);
CREATE TABLE revlog (
  id integer PRIMARY KEY,
  cid integer NOT NULL,
  usn integer NOT NULL,
  ease integer NOT NULL,
  ivl integer NOT NULL,
  lastIvl integer NOT NULL,
  factor integer NOT NULL,
  time integer NOT NULL,
  type integer NOT NULL
);
CREATE TABLE tags (
  tag text NOT NULL PRIMARY KEY COLLATE unicase,
  usn integer NOT NULL,
  collapsed boolean NOT NULL,
  config blob NULL
) without rowid;
CREATE TABLE templates (
  ntid integer NOT NULL,
  ord integer NOT NULL,
  name text NOT NULL COLLATE unicase,
  mtime_secs integer NOT NULL,
  usn integer NOT NULL,
  config blob NOT NULL,
  PRIMARY KEY (ntid, ord)
) without rowid;
CREATE INDEX idx_cards_odid ON cards (odid)
WHERE odid != 0;
CREATE UNIQUE INDEX idx_decks_name ON decks (name);
CREATE UNIQUE INDEX idx_fields_name_ntid ON fields (name, ntid);
CREATE INDEX idx_graves_pending ON graves (usn);
CREATE INDEX idx_notes_mid ON notes (mid);
CREATE UNIQUE INDEX idx_notetypes_name ON notetypes (name);
CREATE INDEX idx_notetypes_usn ON notetypes (usn);
CREATE UNIQUE INDEX idx_templates_name_ntid ON templates (name, ntid);
CREATE INDEX idx_templates_usn ON templates (usn);
CREATE INDEX ix_cards_nid ON cards (nid);
CREATE INDEX ix_cards_sched ON cards (did, queue, due);
CREATE INDEX ix_cards_usn ON cards (usn);
CREATE INDEX ix_notes_csum ON notes (csum);
CREATE INDEX ix_notes_usn ON notes (usn);
CREATE INDEX ix_revlog_cid ON revlog (cid);
CREATE INDEX ix_revlog_usn ON revlog (usn);
"""

def unicase(a:str, b:str):
    "sqlite collation function matching Anki's case-insensitive `unicase`"
    a,b = a.casefold(),b.casefold()
    return -1 if a<b else (1 if a>b else 0)

def connect(path):
    "Connect to collection sqlite file at `path`, with Anki's `unicase` collation and apsw best practices"
    con = apsw.Connection(str(path))
    bestpractice.connection_busy_timeout(con)  # first, so the WAL pragma and every later write wait for the lock rather than erroring
    bestpractice.connection_wal(con)           # WAL journal mode
    bestpractice.connection_dqs(con)           # double quotes are identifiers only; SQL string literals must use single quotes
    con.create_collation('unicase', unicase)
    return con

casefold is not identical to the Unicode case folding in Rust’s unicase crate for every codepoint, but sqlite only consults the collation for ordering and uniqueness checks on names, where casefold agrees for any name you’re likely to meet.

To confirm the DDL still matches what Anki generates, create a fresh collection with the oracle and compare table-by-table:

td = Path(tempfile.mkdtemp())
oc = AnkiCollection(str(td/'collection.anki2'))
oc.close()
ocon = connect(td/'collection.anki2')
theirs = {r[0]: r[1] for r in ocon.execute("select name, sql from sqlite_master where sql is not null and name not like 'sqlite_stat%'")}
mcon = connect(':memory:')
mcon.execute(SCHEMA)
ours = {r[0]: r[1] for r in mcon.execute("select name, sql from sqlite_master where sql is not null")}
test_eq(ours, theirs)

Timestamps and ids

def now_ms(): return int(time.time()*1000)

def day_offset():
    "Minutes west of UTC, Anki's `creationOffset` (eg Australia at +10h is -600)"
    return -int(time.localtime().tm_gmtoff//60)

def creation_stamp():
    "Collection creation time: the most recent 4am in the local timezone"
    t = time.localtime()
    four_am = int(time.mktime((t.tm_year, t.tm_mon, t.tm_mday, 4, 0, 0, 0, 0, -1)))
    return four_am if four_am<=int(time.time()) else four_am-86400

Anki dates everything from crt, the collection creation stamp, pinned to a 4am day rollover so late-night reviews count as the day before. creationOffset records the timezone it was computed in.

Object ids are epoch milliseconds, bumped past any existing id on collision:

def ts_id(con, table):
    "New timestamp-based id for `table`: now in ms, or 1 past the max existing id"
    mx = con.execute(f'select max(id) from {table}').fetchone()[0]
    return max(now_ms(), (mx or 0)+1)

Default objects

A working collection needs one deck (“Default”, id 1), one deck config (id 1), and at least one notetype. We build Basic and Cloze, the two our API creates cards with; anything fancier arrives via sync from collections that already have it. The builders construct the same protobuf messages Anki’s stock-notetype code does, including the precomputed reqs (which fields must be non-empty for each template to generate a card).

CARD_CSS = '.card {\n    font-family: arial;\n    font-size: 20px;\n    line-height: 1.5;\n    text-align: center;\n    color: black;\n    background-color: white;\n}\n'
CLOZE_CSS = CARD_CSS + '.cloze {\n    font-weight: bold;\n    color: blue;\n}\n.nightMode .cloze {\n    color: lightblue;\n}\n'
LATEX_PRE = '\\documentclass[12pt]{article}\n\\special{papersize=3in,5in}\n\\usepackage[utf8]{inputenc}\n\\usepackage{amssymb,amsmath}\n\\pagestyle{empty}\n\\setlength{\\parindent}{0in}\n\\begin{document}\n'
LATEX_POST = '\\end{document}'

def field_cfg(
    tag=None, # Cloze field role marker
    prevent_deletion=False, # Protect the field from deletion in Anki's UI
):
    c = notetypes_pb2.Notetype.Field.Config(font_name='Arial', font_size=20, id=random.getrandbits(63))
    if tag is not None: c.tag = tag
    if prevent_deletion: c.prevent_deletion = True
    return c.SerializeToString()

def template_cfg(q, a):
    c = notetypes_pb2.Notetype.Template.Config(q_format=q, a_format=a, id=random.getrandbits(63))
    return c.SerializeToString()

def notetype_cfg(
    kind=0, # 0 normal, 1 cloze
    stock=0, # original_stock_kind, for Anki's "restore to default" feature
    reqs=None, # [(card_ord, 'any'|'all', [field_ords])] card generation requirements
    css=CARD_CSS,
):
    c = notetypes_pb2.Notetype.Config(kind=kind, css=css, latex_pre=LATEX_PRE, latex_post=LATEX_POST, original_stock_kind=stock)
    for ord_,rkind,fords in reqs or []:
        r = c.reqs.add()
        r.card_ord,r.kind = ord_, notetypes_pb2.Notetype.Config.CardRequirement.Kind.Value(f'KIND_{rkind.upper()}')
        r.field_ords.extend(fords)
    return c.SerializeToString()

Now the two stock notetypes. Each is a set of rows: one in notetypes, one per field in fields, one per template in templates. add_notetype inserts them all:

def add_notetype(con, name, flds, tmpls, cfg):
    "Insert a notetype with `flds` [(name, config_bytes)] and `tmpls` [(name, config_bytes)], returning its id"
    ntid = ts_id(con, 'notetypes')
    con.execute('insert into notetypes values (?,?,?,?,?)', (ntid, name, 0, 0, cfg))
    con.executemany('insert into fields values (?,?,?,?)', [(ntid, i, n, c) for i,(n,c) in enumerate(flds)])
    con.executemany('insert into templates values (?,?,?,?,?,?)', [(ntid, i, n, 0, 0, c) for i,(n,c) in enumerate(tmpls)])
    return ntid

def add_basic(con):
    tmpl = template_cfg('{{Front}}', '{{FrontSide}}\n\n<hr id=answer>\n\n{{Back}}')
    cfg = notetype_cfg(stock=STOCK.Value('ORIGINAL_STOCK_KIND_BASIC'), reqs=[(0,'any',[0])])
    return add_notetype(con, 'Basic', [('Front',field_cfg()),('Back',field_cfg())], [('Card 1',tmpl)], cfg)

def add_cloze(con):
    tmpl = template_cfg('{{cloze:Text}}', '{{cloze:Text}}<br>\n{{Back Extra}}')
    flds = [('Text',field_cfg(tag=0, prevent_deletion=True)),('Back Extra',field_cfg(tag=1))]
    cfg = notetype_cfg(kind=1, css=CLOZE_CSS, stock=STOCK.Value('ORIGINAL_STOCK_KIND_CLOZE'), reqs=[(0,'any',[0])])
    return add_notetype(con, 'Cloze', flds, [('Cloze',tmpl)], cfg)

The test: build ours, decode Anki’s from the oracle collection, and compare as dicts. The random per-field ids (used by newer Ankis to match fields when merging imports) will differ, so they’re dropped before comparing:

add_basic(mcon); _ = add_cloze(mcon)

def ntdicts(con):
    "All notetype/field/template config blobs for Basic and Cloze, decoded, random ids stripped"
    qs = [('notetype', notetypes_pb2.Notetype.Config, "select n.name, 0, n.config from notetypes n"),
          ('field', notetypes_pb2.Notetype.Field.Config, "select n.name, f.ord, f.config from fields f join notetypes n on f.ntid=n.id"),
          ('template', notetypes_pb2.Notetype.Template.Config, "select n.name, t.ord, t.config from templates t join notetypes n on t.ntid=n.id"),]
    res = {}
    for kind,msgcls,q in qs:
        for name,ord_,blob in con.execute(q + " where n.name in ('Basic','Cloze')"):
            m = msgcls(); m.ParseFromString(blob)
            d = MessageToDict(m, preserving_proto_field_name=True); d.pop('id', None)
            res[kind,name,ord_] = d
    return res

test_eq(ntdicts(mcon), ntdicts(ocon))

Creating a collection

create_collection writes everything a fresh collection needs: the col row (legacy JSON columns as empty strings), the config entries Anki expects, the Default deck and deck config, and the two notetypes. The values mirror what Anki 26.05 writes on profile creation, checked by the oracle test at the end.

DEFAULT_CONFIG = dict(activeDecks=[1], addToCur=True, collapseTime=1200, curDeck=1, dayLearnFirst=False,
    dueCounts=True, estTimes=True, newSpread=0, nextPos=1, sched2021=True, schedVer=2,
    sortBackwards=False, sortType='noteFld', timeLim=0)

def default_deck_config():
    c = deck_config_pb2.DeckConfig.Config(learn_steps=[1.0,10.0], relearn_steps=[10.0],
        easy_days_percentages=[1.0]*7, new_per_day=20, reviews_per_day=200, initial_ease=2.5,
        easy_multiplier=1.3, hard_multiplier=1.2, interval_multiplier=1.0, maximum_review_interval=36500,
        minimum_lapse_interval=1, graduating_interval_good=1, graduating_interval_easy=4,
        leech_action=deck_config_pb2.DeckConfig.Config.LEECH_ACTION_TAG_ONLY, leech_threshold=8,
        cap_answer_time_to_secs=60, desired_retention=0.9, historical_retention=0.9, wait_for_audio=True)
    return c.SerializeToString()

def deck_common(): return decks_pb2.Deck.Common(study_collapsed=True, browser_collapsed=True).SerializeToString()

def deck_kind(config_id=1):
    return decks_pb2.Deck.KindContainer(normal=decks_pb2.Deck.Normal(config_id=config_id)).SerializeToString()

def create_collection(path):
    "Create a new empty schema-18 collection at `path`, returning a connection to it"
    path = Path(path)
    assert not path.exists(), f"{path} already exists"
    path.parent.mkdir(parents=True, exist_ok=True)
    con = connect(path)
    con.execute(SCHEMA)
    now = now_ms()
    con.execute("insert into col values (1,?,0,?,18,0,0,0,'','','','','')", (creation_stamp(), now))
    for k,v in DEFAULT_CONFIG.items(): con.execute('insert into config values (?,0,0,?)', (k, json.dumps(v).encode()))
    con.execute('insert into config values (?,0,0,?)', ('creationOffset', json.dumps(day_offset()).encode()))
    con.execute("insert into deck_config values (1,'Default',0,0,?)", (default_deck_config(),))
    con.execute("insert into decks values (1,'Default',0,0,?,?)", (deck_common(), deck_kind()))
    basic = add_basic(con); add_cloze(con)
    con.execute('insert into config values (?,-1,?,?)', ('curModel', now//1000, json.dumps(basic).encode()))
    return con

The moment of truth: Anki opens a file we wrote from nothing, passes its own database check, and adds a note to it.

cpath = td/'ours'/'collection.anki2'
con = create_collection(cpath)
con.close()
ac = AnkiCollection(str(cpath))
prob = ac.fix_integrity()
test_eq(prob[1], True)
prob

nt = ac.models.by_name('Basic')
n = ac.new_note(nt)
n['Front'],n['Back'] = 'hello','world'
ac.add_note(n, 1)
ac.close()
con = connect(cpath)
flds, = con.execute('select flds from notes').fetchone()
test_eq(flds, 'hello\x1fworld')
test_eq(con.execute('select count(*) from cards').fetchone()[0], 1)

Beyond passing fix_integrity, the values a fresh collection is born with match Anki’s own. The default deck and its config decode to identical protobufs, and the config table agrees key for key — excepting curModel, which names the Basic notetype by its per-collection timestamp id:

ourcon = create_collection(td/'cmp.anki2')
def deckblob(c):
    cm,kd = c.execute('select common, kind from decks where id=1').fetchone()
    return (MessageToDict(decks_pb2.Deck.Common.FromString(cm), preserving_proto_field_name=True),
            MessageToDict(decks_pb2.Deck.KindContainer.FromString(kd), preserving_proto_field_name=True))
def dconfblob(c):
    b, = c.execute('select config from deck_config where id=1').fetchone()
    return MessageToDict(deck_config_pb2.DeckConfig.Config.FromString(b), preserving_proto_field_name=True)
def conftbl(c): return {k:json.loads(v) for k,v in c.execute('select key, val from config') if k!='curModel'}
test_eq(deckblob(ourcon), deckblob(ocon))
test_eq(dconfblob(ourcon), dconfblob(ocon))
test_eq(conftbl(ourcon), conftbl(ocon))
ourcon.close()