Collections

Notes, cards, decks and search, straight to the sqlite file

With fastanki.schema handling the file format, this module does the actual work: a Collection class that adds, finds, updates and removes notes, generating cards the way Anki would. Every change is bookkept for sync (usn=-1 marks pending rows, deletions go to graves, col.mod tracks modification time), so a collection managed here syncs cleanly with AnkiWeb and any other Anki client.

import hashlib, html, random, time, json
from contextlib import contextmanager
from fastcore.utils import *
from fastanki.schema import *
from fastanki._proto import notetypes_pb2

import tempfile
from fastcore.test import *
from anki.collection import Collection as AnkiCollection
import anki.lang
from anki.utils import field_checksum as anki_csum
from unittest.mock import patch as mock_patch

Text utilities

Anki fingerprints the first field of every note with a checksum used for duplicate detection, computed over the text with HTML stripped but media filenames kept (so <img src=cat.jpg> and a note that mentions cat.jpg collide, as they should). The guid identifies a note across collections, as 10 or so characters of base91.

_media_re = re.compile(r'(?i)<(?:img|audio|object)\b[^>]*?\bsrc\s*=\s*(?:"([^"]*)"|\'([^\']*)\'|([^\s>]+))[^>]*>')
_html_re = re.compile(r'(?is)<style.*?(?:</style>|$)|<script.*?(?:</script>|$)|<!--.*?-->|<.*?>')

def strip_html_media(s:str):
    "Strip HTML from `s`, preserving media filenames, matching Anki's checksum preprocessing"
    s = _media_re.sub(lambda m: f" {first(g for g in m.groups() if g is not None)} ", s)
    s = _html_re.sub('', s)
    return html.unescape(s).replace('\xa0',' ')

def field_csum(text:str):
    "Anki's dupe-detection checksum: first 4 bytes of the sha1 of the stripped first field"
    return int.from_bytes(hashlib.sha1(strip_html_media(text).encode()).digest()[:4], 'big')

_B91 = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!#$%&()*+,-./:;<=>?@[]^_`{|}~"

def guid64():
    "A new random note guid, in Anki's base91 encoding"
    n = random.getrandbits(64)
    buf = ''
    while True:                                         # do-while: a base91 number always has >=1 digit, so the guid is never empty
        n,r = divmod(n, len(_B91)); buf += chr(_B91[r])
        if not n: return buf

Checked directly against Anki’s own implementation, on plain text, HTML, entities, and media references:

anki.lang.set_lang('en')  # pylib's checksum helper strips HTML through a backend
for s in ['hello', 'dos', '<b>test</b>', 'a &amp; b', '<img src=cat.jpg>', "<img src='dog gif.gif'>x",
          '[sound:beep.mp3]', 'caf\xe9&nbsp;au lait', '<style>p {}</style>text']:
    test_eq(field_csum(s), anki_csum(s))

test_eq(field_csum('hello'), 2868168221)
g = guid64()
assert 1 <= len(g) <= 11 and set(g) <= set(map(chr, _B91))
g

Opening a collection

fastanki keeps its own collection, separate from any desktop Anki profile: on machines where fastanki runs there’s usually no desktop Anki at all, and where there is one, staying out of its data folder means never fighting it for the sqlite lock. The desktop sees our changes the same way your phone does, through AnkiWeb sync.

def data_dir():
    "fastanki's data folder (override with FASTANKI_DIR)"
    return Path(os.environ.get('FASTANKI_DIR', '~/.fastanki')).expanduser()

class Collection:
    def __init__(self, path): self.path,self.con = Path(path),connect(path)
    @classmethod
    def open(cls, path=None):
        "Open collection at `path` (default `data_dir()/collection.anki2`), creating if needed"
        path = Path(path) if path else data_dir()/'collection.anki2'
        if not path.exists(): create_collection(path).close()
        return cls(path)
    def close(self): self.con.close()
    def __enter__(self): return self
    def __exit__(self, *args): self.close()
    def q(self, sql, *params): return self.con.execute(sql, params).fetchall()
    def q1(self, sql, *params):
        "Single value from a single row, or None"
        r = self.con.execute(sql, params).fetchone()
        return r[0] if r else None
    def _dirty(self):
        "Record that the collection changed, for sync's modification check"
        # max(...,mod+1) keeps mod strictly increasing: a change in the same ms as the last sync's finish must still look newer, else meta reports 'noChanges' and skips it
        self.con.execute('update col set mod=max(?, mod+1)', (now_ms(),))
    @contextmanager
    def _tx(self):
        "One write transaction so id allocation and writes stay atomic across connections: BEGIN IMMEDIATE at the top level, nested calls reuse it via a savepoint"
        if self.con.getautocommit():
            self.con.execute('BEGIN IMMEDIATE')
            try: yield; self.con.execute('COMMIT')
            except: self.con.execute('ROLLBACK'); raise
        else:
            with self.con: yield

q and q1 keep the sql one-liners readable throughout the rest of the module. _dirty bumps col.mod, which is how sync notices there’s something to send.

td = Path(tempfile.mkdtemp())
col = Collection.open(td/'collection.anki2')
test_eq(col.q1('select ver from col'), 18)

Notetypes

A notetype’s structure spans three tables plus a protobuf config. NT gathers what note operations need: field names, sort field, kind, and the per-template card generation requirements.

class NT:
    "A notetype's structure: fields, templates, and card generation rules"
    def __init__(self, con, ntid, name):
        self.id,self.name = ntid,name
        cfg = notetypes_pb2.Notetype.Config()
        cfg.ParseFromString(con.execute('select config from notetypes where id=?',(ntid,)).fetchone()[0])
        self.cloze = cfg.kind==1
        self.sortf = cfg.sort_field_idx
        self.flds = [r[0] for r in con.execute('select name from fields where ntid=? order by ord',(ntid,))]
        self.tmpls = []   # (ord, front-template) per card template, for render-based generation
        for o,b in con.execute('select ord, config from templates where ntid=? order by ord',(ntid,)):
            tc = notetypes_pb2.Notetype.Template.Config(); tc.ParseFromString(b); self.tmpls.append((o, tc.q_format))
    def __repr__(self): return f"NT({self.name!r}, flds={self.flds}, {'cloze' if self.cloze else f'{len(self.tmpls)} templates'})"

@patch
def nt(self:Collection, name):
    "Notetype structure by name (case-insensitive)"
    r = self.con.execute('select id, name from notetypes where name=?',(name,)).fetchone()
    if not r: raise KeyError(f"No notetype {name!r}")
    return NT(self.con, *r)

@patch
def notetypes(self:Collection): return [r[0] for r in self.q('select name from notetypes order by name')]

col.nt('Basic')

The name=? lookup goes through sqlite’s unicase collation on the name column, so col.nt('basic') finds Basic:

test_eq(col.nt('basic').id, col.nt('Basic').id)
test_eq(col.notetypes(), ['Basic','Cloze'])

Decks

Deck names nest with :: in Anki’s UI, but the table stores the separator as \x1f, one row per level. Adding Spanish::Vocab creates Spanish too, as Anki does.

def _dname(name): return name.replace('::','\x1f')

@patch
def deck_id(self:Collection, name, create=False):
    "Id of deck `name` (`::`-separated), optionally creating it (and any missing parents)"
    did = self.q1('select id from decks where name=?', _dname(name))
    if did or not create: return did
    parts = name.split('::')
    with self._tx():
        for i in range(1, len(parts)):
            self.deck_id('::'.join(parts[:i]), create=True)
        did = ts_id(self.con, 'decks')
        self.con.execute('insert into decks values (?,?,?,-1,?,?)', (did, _dname(name), now_ms()//1000, deck_common(), deck_kind()))
        self._dirty()
    return did

@patch
def decks(self:Collection): return [r[0].replace('\x1f','::') for r in self.q('select name from decks order by name')]

did = col.deck_id('Spanish::Vocab', create=True)
test_eq(col.decks(), ['Default','Spanish','Spanish::Vocab'])
test_eq(col.deck_id('Spanish::Vocab'), did)
test_is(col.deck_id('Nope'), None)

Adding notes

Note is a plain holder for one row of the notes table, with dict-style field access by name. The interesting work is in add_note: joining fields with \x1f, computing sfld and csum, and deciding which cards to generate. For normal notetypes each template carries a requirement (“any of these fields non-empty” or “all of them”); for cloze notetypes there’s one card per distinct {c1::...} number across the fields.

Card generation follows Anki’s rule rather than the stored req cache: a normal template generates a card when its front would render non-empty for the note’s non-empty fields. renders_with_fields parses the front template ({Field}, {#Field}…{{/Field}}, {^Field}…{{/Field}}, comments) and reports whether any non-empty field reaches the output; static text alone never counts. field_is_empty matches Anki: whitespace and empty <br>/<div> tags only.

_field_empty_re = re.compile(r'(?si)^(?:\s|</?(?:br|div) ?/?>)*$')
def field_is_empty(s): return bool(_field_empty_re.match(s))

_tmpl_re = re.compile(r'\{\{(.+?)\}\}', re.S)
def _parse_tmpl(qfmt):
    "Parse a template into nested nodes: ('text',) ('rep',key) ('cond',key,kids) ('negc',key,kids)"
    toks,pos = [],0
    for m in _tmpl_re.finditer(qfmt):
        if m.start()>pos: toks.append(('text',))
        c = m.group(1).strip()
        if   c.startswith('#'): toks.append(('open', c[1:].strip()))
        elif c.startswith('^'): toks.append(('neg',  c[1:].strip()))
        elif c.startswith('/'): toks.append(('close',))
        elif c.startswith('!'): toks.append(('text',))
        else: toks.append(('rep', c.rsplit(':',1)[-1].strip('{} ')))
        pos = m.end()
    if pos<len(qfmt): toks.append(('text',))
    def build(i):
        nodes=[]
        while i<len(toks):
            t=toks[i]
            if t[0]=='close': return nodes,i+1
            if t[0] in ('open','neg'): kids,i=build(i+1); nodes.append(('cond' if t[0]=='open' else 'negc', t[1], kids))
            else: nodes.append(t); i+=1
        return nodes,i
    return build(0)[0]

def _tmpl_empty(nodes, nonempty):
    "Anki's rule: a template is empty unless some non-empty field's content reaches the output"
    for n in nodes:
        if n[0]=='rep' and n[1] in nonempty: return False
        if n[0]=='cond' and n[1] in nonempty and not _tmpl_empty(n[2], nonempty): return False
        if n[0]=='negc' and n[1] not in nonempty and not _tmpl_empty(n[2], nonempty): return False
    return True

def renders_with_fields(qfmt, nonempty):
    "Would front template `qfmt` render non-empty given the set of `nonempty` field names?"
    return not _tmpl_empty(_parse_tmpl(qfmt), nonempty)

class Note:
    def __init__(self, id, guid, mid, mod, usn, tags, flds, nt):
        store_attr()
        self.fields = dict(zip(nt.flds, flds.split('\x1f')))
        self.tags = tags.split()
    def __getitem__(self, k): return self.fields[k]
    def __setitem__(self, k, v):
        if k not in self.fields: raise KeyError(f"No field {k!r} in {self.nt.name}")
        self.fields[k] = v
    def __repr__(self):
        flds = ', '.join(f'{k}={v!r}' for k,v in self.fields.items())
        return f"Note({self.id}, {flds}, tags={self.tags})"
    def _repr_markdown_(self):
        flds = ' | '.join(f"**{k}**: {v}" for k,v in self.fields.items())
        return f"{flds} | \N{LABEL} {' '.join(self.tags) or '-'}"

_cloze_re = re.compile(r'\{\{c(\d+)::')

_SPECIAL_FIELDS = ('Card','CardFlag','Deck','Subdeck','Type','CardID')  # always-present special fields (FrontSide excluded on the front; Tags handled below)
def note_cards(nt, fields, tags=()):
    "Template ordinals that generate cards for a note's `fields` (values in field order) and `tags`, using Anki's render rule"
    if nt.cloze: return sorted({min(int(m)-1,499) for f in fields for m in _cloze_re.findall(f) if int(m)>=1})
    nonempty = {f for f,v in zip(nt.flds, fields) if not field_is_empty(v)}
    nonempty |= {s for s in _SPECIAL_FIELDS if s not in nt.flds}
    if tags and 'Tags' not in nt.flds: nonempty.add('Tags')
    return [o for o,qfmt in nt.tmpls if renders_with_fields(qfmt, nonempty)]

Cloze ordinals are 1-based in note text ({c1::...}) but 0-based as card ords, so note_cards returns them shifted. KIND_NONE (0) requirements mean the template never generates; any is 1 and all is 2 in the protobuf enum.

bnt = col.nt('Basic')
test_eq(note_cards(bnt, ['hi','']), [0])
test_eq(note_cards(bnt, ['','back only']), [])
cnt = col.nt('Cloze')
test_eq(note_cards(cnt, ['{{c1::a}} and {{c3::b}}','']), [0,2])
test_eq(note_cards(cnt, ['no clozes here','']), [])

And the render rule agrees with Anki itself across the reversed and optional-reversed (conditional {#Add Reverse}) notetypes: Anki generates the cards, and note_cards predicts the same ordinals from the note’s fields.

gp = Path(tempfile.mkdtemp())/'gen.anki2'
aoc = AnkiCollection(str(gp))
want = []
for mname, flds in [('Basic (and reversed card)', dict(Front='q', Back='a')),
                    ('Basic (and reversed card)', dict(Front='q', Back='')),
                    ('Basic (optional reversed card)', dict(Front='q', Back='a')),
                    ('Basic (optional reversed card)', dict(Front='q', Back='a', **{'Add Reverse':'y'}))]:
    n = aoc.new_note(aoc.models.by_name(mname))
    for k,v in flds.items(): n[k] = v
    aoc.add_note(n, 1)
    want.append((mname, flds, sorted(c.ord for c in n.cards())))
aoc.close()
fc = Collection(gp)
for mname, flds, ords in want: test_eq(note_cards(fc.nt(mname), [flds.get(f,'') for f in fc.nt(mname).flds]), ords)
fc.close()

Now add itself. Anki assigns each new card a position (due for a new card is its queue position, not a date), drawn from the nextPos counter:

@patch
def _next_pos(self:Collection):
    pos = json.loads(self.q1("select val from config where key='nextPos'") or b'0')
    self.con.execute("insert or replace into config values ('nextPos',-1,?,?)", (now_ms()//1000, json.dumps(pos+1).encode()))
    return pos

@patch
def add(self:Collection, model='Basic', deck='Default', tags=None, **fields):
    "Add a note (and its cards), returning the `Note`"
    nt = self.nt(model)
    unknown = set(fields) - set(nt.flds)
    if unknown: raise KeyError(f"Fields not in {nt.name}: {unknown}")
    vals = [fields.get(f,'') for f in nt.flds]
    ords = note_cards(nt, vals, tags)
    if not ords: raise ValueError("Note would generate no cards (check required fields)")
    with self._tx():
        did = self.deck_id(deck, create=True)
        nid,now = ts_id(self.con,'notes'), now_ms()//1000
        tags = [tags] if isinstance(tags,str) else (tags or [])
        tagstr = f" {' '.join(tags)} " if tags else ''
        self.con.execute("insert into notes values (?,?,?,?,-1,?,?,?,?,0,'')",
            (nid, guid64(), nt.id, now, tagstr, '\x1f'.join(vals), strip_html_media(vals[nt.sortf]), field_csum(vals[0])))
        cid = ts_id(self.con,'cards')
        for i,o in enumerate(ords):
            self.con.execute("insert into cards values (?,?,?,?,?,-1,0,0,?,0,0,0,0,0,0,0,0,'{}')",
                (cid+i, nid, did, o, now, self._next_pos()))
        self._dirty()
    return self.get_note(nid)

@patch
def get_note(self:Collection, nid):
    "Retrieve a `Note` by id"
    r = self.con.execute('select id, guid, mid, mod, usn, tags, flds from notes where id=?',(nid,)).fetchone()
    if not r: raise KeyError(f"No note {nid}")
    ntr = self.con.execute('select id, name from notetypes where id=?',(r[2],)).fetchone()
    return Note(*r, nt=NT(self.con, *ntr))

n = col.add(Front='hola', Back='hello', deck='Spanish::Vocab', tags='spanish')
n

test_eq(n['Front'], 'hola')
test_eq(n.tags, ['spanish'])
test_eq(col.q1('select count(*) from cards where nid=?', n.id), 1)
test_eq(col.q1('select due from cards where nid=?', n.id), 1)
cz = col.add(model='Cloze', Text='{{c1::uno}} y {{c2::dos}}')
test_eq(col.q('select ord, due from cards where nid=? order by ord', cz.id), [(0,2),(1,3)])

Updating and removing

Updates rewrite the note row with fresh mod, usn=-1, sfld and csum, and generate any cards a new cloze number now requires (never deleting existing ones, matching Anki). Removals record graves, the tombstones sync uses to propagate deletions: type 0 is a card, 1 a note, 2 a deck.

@patch
def update_note(self:Collection, note, tags=None, add_tags=None, **fields):
    "Update fields and/or tags of `note` (a `Note` or note id), returning the updated `Note`"
    if not isinstance(note,Note): note = self.get_note(note)
    for k,v in fields.items(): note[k] = v
    if tags is not None: note.tags = [tags] if isinstance(tags,str) else tags
    if add_tags: note.tags += [t for t in ([add_tags] if isinstance(add_tags,str) else add_tags) if t not in note.tags]
    vals = [note.fields[f] for f in note.nt.flds]
    tagstr = f" {' '.join(note.tags)} " if note.tags else ''
    with self._tx():
        self.con.execute('update notes set mod=?, usn=-1, tags=?, flds=?, sfld=?, csum=? where id=?',
            (now_ms()//1000, tagstr, '\x1f'.join(vals), strip_html_media(vals[note.nt.sortf]), field_csum(vals[0]), note.id))
        have = {r[0] for r in self.q('select ord from cards where nid=?', note.id)}
        did = self.q1('select did from cards where nid=?', note.id)
        cid,now = ts_id(self.con,'cards'), now_ms()//1000
        for i,o in enumerate(o for o in note_cards(note.nt, vals, note.tags) if o not in have):
            self.con.execute("insert into cards values (?,?,?,?,?,-1,0,0,?,0,0,0,0,0,0,0,0,'{}')",
                (cid+i, note.id, did, o, now, self._next_pos()))
        self._dirty()
    return self.get_note(note.id)

n2 = col.update_note(n, Back='hello!', add_tags='greeting')
test_eq(n2['Back'], 'hello!')
test_eq(n2.tags, ['spanish','greeting'])
cz2 = col.update_note(cz.id, Text='{{c1::uno}} {{c2::dos}} {{c3::tres}}')
test_eq(col.q1('select count(*) from cards where nid=?', cz.id), 3)

@patch
def remove_notes(self:Collection, nids):
    "Remove notes (and their cards) by id, recording graves for sync"
    nids = [n.id if isinstance(n,Note) else n for n in listify(nids)]
    with self._tx():
        for nid in nids:
            for (cid,) in self.q('select id from cards where nid=?', nid):
                self.con.execute('insert or ignore into graves values (?,0,-1)', (cid,))
            self.con.execute('insert or ignore into graves values (?,1,-1)', (nid,))
            self.con.execute('delete from cards where nid=?', (nid,))
            self.con.execute('delete from notes where id=?', (nid,))
        self._dirty()
    return len(nids)

@patch
def remove_deck(self:Collection, name):
    "Remove deck `name` and its subdecks, with their cards and notes"
    pat = _dname(name)
    dids = [r[0] for r in self.q("select id from decks where name=? or name like ?", pat, pat+'\x1f%')]
    if not dids: return 0
    ph = ','.join('?'*len(dids))
    nids = [r[0] for r in self.q(f'select distinct nid from cards where did in ({ph})', *dids)]
    with self._tx():
        self.remove_notes(nids)
        for did in dids:
            self.con.execute('insert or ignore into graves values (?,2,-1)', (did,))
            self.con.execute('delete from decks where id=?', (did,))
        self._dirty()
    return len(dids)

Anki’s “remove deck” takes the notes down with it when their cards all lived there; we do the simpler, stricter thing and remove every note that had a card in the deck. The functional API only exposes deck removal through Collection, so surprises stay contained.

col.add(Front='bye', deck='Doomed')
test_eq(col.remove_deck('Doomed'), 1)
test_is(col.deck_id('Doomed'), None)
test_eq(col.q1("select count(*) from notes where flds like 'bye%'"), 0)
test_eq(col.q1("select count(*) from graves where type=2"), 1)

Finding

Anki has a search language; we have Python. Keyword arguments compose to SQL: deck, tag and added_days map to indexed columns, any other keyword is treated as a field name and matched case-insensitively as a substring, and where/args drop through to raw SQL against the notes table (aliased n) joined with cards (aliased c) for anything the keywords don’t cover.

@patch
def _find_sql(self:Collection, deck=None, tag=None, added_days=None, where=None, args=()):
    conds,ps = [],[]
    if deck:
        pat = _dname(deck)
        conds.append('c.did in (select id from decks where name=? or name like ?)'); ps += [pat, pat+'\x1f%']
    if tag: conds.append('n.tags like ?'); ps.append(f'% {tag} %')
    if added_days is not None: conds.append('n.id >= ?'); ps.append((int(time.time())-added_days*86400)*1000)
    if where: conds.append(f'({where})'); ps += list(args)
    return ' and '.join(conds) or '1=1', ps

def _fld_match(nt_flds, fields, flds_str):
    d = dict(zip(nt_flds, flds_str.split('\x1f')))
    return all(k in d and v.casefold() in d[k].casefold() for k,v in fields.items())

@patch
def find_notes(self:Collection, deck=None, tag=None, added_days=None, where=None, args=(), **fields):
    "Notes matching all given criteria; field kwargs are case-insensitive substring matches"
    cond,ps = self._find_sql(deck, tag, added_days, where, args)
    sql = f'select distinct n.id, n.guid, n.mid, n.mod, n.usn, n.tags, n.flds from notes n join cards c on c.nid=n.id where {cond}'
    nts,res = {},[]
    for r in self.q(sql, *ps):
        mid = r[2]
        if mid not in nts: nts[mid] = NT(self.con, mid, self.q1('select name from notetypes where id=?', mid))
        if not fields or _fld_match(nts[mid].flds, fields, r[6]): res.append(Note(*r, nt=nts[mid]))
    return sorted(res, key=lambda n:n.id)

@patch
def find_note_ids(self:Collection, **kw): return [n.id for n in self.find_notes(**kw)]

test_eq(col.find_note_ids(deck='Spanish'), [n.id])
test_eq(col.find_note_ids(tag='greeting'), [n.id])
test_eq([x.id for x in col.find_notes(Front='HOLA')], [n.id])
test_eq(col.find_notes(Front='nope'), [])
test_eq(len(col.find_notes()), 2)
test_eq(col.find_note_ids(where='n.id=?', args=[cz.id]), [cz.id])

Cards get the same treatment. Card carries the scheduling columns; is_due selects learning cards whose time has come and review cards due today or earlier.

class Card:
    def __init__(self, id, nid, did, ord, mod, usn, type, queue, due, ivl): store_attr()
    def __repr__(self): return f"Card({self.id}, nid={self.nid}, due={self.due}, ivl={self.ivl}, queue={self.queue})"
    def _repr_markdown_(self): return f"Card {self.id} (nid: {self.nid}, due: {self.due}, ivl: {self.ivl}d, queue: {self.queue})"

@patch
def today(self:Collection):
    "Days since collection creation, Anki's day counter"
    return (int(time.time()) - self.q1('select crt from col'))//86400

@patch
def find_cards(self:Collection, deck=None, tag=None, added_days=None, is_due=None, where=None, args=(), **fields):
    "Cards matching all given criteria (same keywords as `find_notes`, plus `is_due`)"
    nids = None
    if fields: nids = {x.id for x in self.find_notes(deck=deck, tag=tag, added_days=added_days, **fields)}
    cond,ps = self._find_sql(deck, tag, added_days, where, args)
    if is_due: cond += f' and (c.queue=1 and c.due<=? or c.queue in (2,3) and c.due<=?)'; ps += [int(time.time())+1200, self.today()]
    sql = f'select c.id, c.nid, c.did, c.ord, c.mod, c.usn, c.type, c.queue, c.due, c.ivl from cards c join notes n on c.nid=n.id where {cond} order by c.id'
    return [Card(*r) for r in self.q(sql, *ps) if nids is None or r[1] in nids]

@patch
def find_card_ids(self:Collection, **kw): return [c.id for c in self.find_cards(**kw)]

cards = col.find_cards(deck='Spanish')
test_eq(len(cards), 1)
cards[0]

New cards aren’t “due” in Anki’s sense, and nothing has been studied in this collection, so:

test_eq(col.find_cards(is_due=True), [])
test_eq(len(col.find_card_ids()), 4)

Due counts

@patch
def due_counts(self:Collection, deck=None):
    "(new, learning, review) counts for `deck` and its subdecks (or the whole collection)"
    cond,ps = self._find_sql(deck)
    sql = ('select sum(c.queue=0), sum(c.queue=1 and c.due<=? or c.queue=3 and c.due<=?), sum(c.queue=2 and c.due<=?) '
           f'from cards c join notes n on c.nid=n.id where {cond}')
    r = self.q(sql, int(time.time())+1200, self.today(), self.today(), *ps)[0]
    return tuple(x or 0 for x in r)

test_eq(col.due_counts(), (4,0,0))
test_eq(col.due_counts('Spanish'), (1,0,0))

The oracle test

Everything above wrote rows we believe are Anki-compatible. Time to ask Anki: open the collection we’ve been mutating, run its database check, use its own search to find our notes, add a note of its own, and make sure we can read that back.

col.close()
ac = AnkiCollection(str(td/'collection.anki2'))
report, ok = ac.fix_integrity()
test_eq(ok, True)

test_eq(ac.find_notes('deck:Spanish::Vocab'), [n.id])
test_eq(ac.find_notes('tag:greeting'), [n.id])
test_eq(sorted(ac.find_notes('')), sorted([n.id, cz.id]))
anote = ac.get_note(cz.id)
test_eq(anote['Text'], '{{c1::uno}} {{c2::dos}} {{c3::tres}}')

theirs = ac.new_note(ac.models.by_name('Basic'))
theirs['Front'] = 'from anki'
ac.add_note(theirs, 1)
ac.close()
col = Collection.open(td/'collection.anki2')
test_eq(col.find_note_ids(Front='from anki'), [theirs.id])
col.close()