Collection files

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

This module creates and opens Anki’s schema-18 .anki2 files without Anki or its Rust library. Each collection is a sqlite database containing notes, cards and review history. Notetype, deck and deck-config settings use protobuf messages stored in database columns. fastanki._proto contains Python classes generated from Anki’s .proto files.

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

The tests use the anki package to check compatibility. It is a development dependency only.

The schema

The schema comes from a collection created by Anki 26.05. Schema 18 retains the old col row with empty strings in its JSON columns. Separate tables now store that data.

Name comparisons require Anki’s case-insensitive unicase collation. connect registers it with sqlite before running any queries.

connect

def connect(
    path
):

Connect to collection sqlite file at path, with Anki’s unicase collation and apsw best practices

unicase

def unicase(
    a:str, b:str
):

sqlite collation function matching Anki’s case-insensitive unicase

Python’s casefold differs from Rust’s unicase for some codepoints. This affects name ordering and uniqueness checks.

Compare the schema with a fresh Anki collection:

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

creation_stamp

def creation_stamp():

Collection creation time: the most recent 4am in the local timezone

day_offset

def day_offset():

Minutes west of UTC, Anki’s creationOffset (eg Australia at +10h is -600)

now_ms

def now_ms():

crt records the collection’s creation time at the most recent local 4am. Anki’s day starts at 4am, counting late-night reviews toward the previous day. creationOffset records the timezone offset in minutes west of UTC.

Object ids use epoch milliseconds. ts_id returns at least one more than the table’s highest existing id.

ts_id

def ts_id(
    con, table
):

New timestamp-based id for table: now in ms, or 1 past the max existing id

Default objects

A new collection contains the Default deck and deck config, both with id 1. It also needs a notetype. fastanki creates Basic and Cloze notetypes with Anki’s defaults. Other notetypes come from existing collections through sync.

The reqs settings specify which fields must be non-empty for each template to generate a card.

notetype_cfg

def notetype_cfg(
    kind:int=0, # 0 normal, 1 cloze
    stock:int=0, # original_stock_kind, for Anki's "restore to default" feature
    reqs:NoneType=None, # [(card_ord, 'any'|'all', [field_ords])] card generation requirements
    css:str='.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'
):

template_cfg

def template_cfg(
    q, a
):

field_cfg

def field_cfg(
    tag:NoneType=None, # Cloze field role marker
    prevent_deletion:bool=False, # Protect the field from deletion in Anki's UI
):

add_notetype inserts the notetype and its fields and templates into their respective tables.

add_cloze

def add_cloze(
    con
):

add_basic

def add_basic(
    con
):

add_notetype

def add_notetype(
    con, name, flds, tmpls, cfg
):

Insert a notetype with flds [(name, config_bytes)] and tmpls [(name, config_bytes)], returning its id

Compare Basic and Cloze with Anki’s versions. Exclude the random ids that Anki uses to match fields during import.

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 initializes the database with these defaults and Anki’s collection settings.

deck_kind

def deck_kind(
    config_id:int=1
):

deck_common

def deck_common():

default_deck_config

def default_deck_config():

create_collection

def create_collection(
    path
):

Create a new empty schema-18 collection at path, returning a connection to it

Check that Anki opens the collection, passes its database check, and adds a note:

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

('Database rebuilt and optimized.', True)

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)

Compare the default deck, deck config and collection settings with Anki. Exclude curModel, which contains the Basic notetype’s per-collection 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()