Skip to content

donna.skills.database

donna.skills.database

Lightweight read-only DB access for skills.

SkillDatabase 

SkillDatabase(connection: Connection)
Source code in src/donna/skills/database.py 
18
19
def __init__(self, connection: aiosqlite.Connection) -> None:
    self._conn = connection

list_all async 

list_all(state: str | None = None, limit: int = 200) -> list[SkillRow]
Source code in src/donna/skills/database.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
async def list_all(self, state: str | None = None, limit: int = 200) -> list[SkillRow]:
    if state:
        cursor = await self._conn.execute(
            f"SELECT {SELECT_SKILL} FROM skill WHERE state = ? "
            "ORDER BY updated_at DESC LIMIT ?",
            (state, limit),
        )
    else:
        cursor = await self._conn.execute(
            f"SELECT {SELECT_SKILL} FROM skill ORDER BY updated_at DESC LIMIT ?",
            (limit,),
        )
    rows = await cursor.fetchall()
    return [row_to_skill(r) for r in rows]

get_by_id async 

get_by_id(skill_id: str) -> SkillRow | None
Source code in src/donna/skills/database.py 
36
37
38
39
40
41
42
async def get_by_id(self, skill_id: str) -> SkillRow | None:
    cursor = await self._conn.execute(
        f"SELECT {SELECT_SKILL} FROM skill WHERE id = ?",
        (skill_id,),
    )
    row = await cursor.fetchone()
    return row_to_skill(row) if row else None

get_by_capability async 

get_by_capability(capability_name: str) -> SkillRow | None
Source code in src/donna/skills/database.py 
44
45
46
47
48
49
50
async def get_by_capability(self, capability_name: str) -> SkillRow | None:
    cursor = await self._conn.execute(
        f"SELECT {SELECT_SKILL} FROM skill WHERE capability_name = ?",
        (capability_name,),
    )
    row = await cursor.fetchone()
    return row_to_skill(row) if row else None

get_version async 

get_version(version_id: str) -> SkillVersionRow | None
Source code in src/donna/skills/database.py 
52
53
54
55
56
57
58
async def get_version(self, version_id: str) -> SkillVersionRow | None:
    cursor = await self._conn.execute(
        f"SELECT {SELECT_SKILL_VERSION} FROM skill_version WHERE id = ?",
        (version_id,),
    )
    row = await cursor.fetchone()
    return row_to_skill_version(row) if row else None