room_cal/room_cal/store/_sqlite.py

import logging
from datetime import datetime
from pathlib import Path

from room_cal.store._backend import EventRecord, register_backend

_log = logging.getLogger(__name__)

_CREATE_TABLE = '''
CREATE TABLE IF NOT EXISTS events (
  uid           TEXT NOT NULL,
  calendar_name TEXT NOT NULL,
  start         TEXT NOT NULL,
  end           TEXT NOT NULL,
  all_day       INTEGER NOT NULL,
  title         TEXT NOT NULL,
  description   TEXT NOT NULL DEFAULT '',
  x_data        TEXT,
  PRIMARY KEY (uid, calendar_name)
)
'''

_COLS = 'uid, calendar_name, start, end, all_day, title, description, x_data'


@register_backend('sqlite')
class SQLiteBackend:
  def __init__(self, db_path: Path, calendar_name: str):
    self._db_path = db_path
    self._calendar_name = calendar_name

  @classmethod
  def from_store_config(cls, store_cfg, calendar_name: str) -> 'SQLiteBackend':
    from room_cal._utils import data_home
    db_path_str = store_cfg.sqlite.db_path
    db_path = Path(db_path_str) if db_path_str else data_home() / 'store.db'
    return cls(db_path, calendar_name)

  def _to_record(self, row) -> EventRecord:
    uid, _, start, end, all_day, title, description, x_data = row
    return EventRecord(
      uid=uid,
      start=datetime.fromisoformat(start),
      end=datetime.fromisoformat(end),
      all_day=bool(all_day),
      title=title,
      description=description,
      x_data=x_data,
    )

  async def fetch_range(self, start: datetime, end: datetime) -> list[EventRecord]:
    import aiosqlite
    async with aiosqlite.connect(self._db_path) as db:
      await db.execute(_CREATE_TABLE)
      cursor = await db.execute(
        f'SELECT {_COLS} FROM events WHERE calendar_name = ? AND start < ? AND end > ? ORDER BY start',
        (self._calendar_name, end.isoformat(), start.isoformat()),
      )
      rows = await cursor.fetchall()
    return [self._to_record(r) for r in rows]

  async def create(self, record: EventRecord) -> EventRecord:
    import aiosqlite
    async with aiosqlite.connect(self._db_path) as db:
      await db.execute(_CREATE_TABLE)
      await db.execute(
        f'INSERT INTO events ({_COLS}) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
        (record.uid, self._calendar_name, record.start.isoformat(), record.end.isoformat(),
         int(record.all_day), record.title, record.description, record.x_data),
      )
      await db.commit()
    _log.info('create date=%s title=%r', record.start.date(), record.title)
    return record

  async def update(self, uid: str, *, title: str | None = None,
                   description: str | None = None, x_data: str | None = None) -> EventRecord:
    import aiosqlite
    async with aiosqlite.connect(self._db_path) as db:
      await db.execute(_CREATE_TABLE)
      if title is not None:
        await db.execute(
          'UPDATE events SET title = ? WHERE uid = ? AND calendar_name = ?',
          (title, uid, self._calendar_name),
        )
      if description is not None:
        await db.execute(
          'UPDATE events SET description = ? WHERE uid = ? AND calendar_name = ?',
          (description, uid, self._calendar_name),
        )
      if x_data is not None:
        await db.execute(
          'UPDATE events SET x_data = ? WHERE uid = ? AND calendar_name = ?',
          (x_data, uid, self._calendar_name),
        )
      await db.commit()
      cursor = await db.execute(
        f'SELECT {_COLS} FROM events WHERE uid = ? AND calendar_name = ?',
        (uid, self._calendar_name),
      )
      row = await cursor.fetchone()
    _log.info('update uid=%s', uid)
    return self._to_record(row)

  async def delete(self, uid: str) -> None:
    import aiosqlite
    async with aiosqlite.connect(self._db_path) as db:
      await db.execute(_CREATE_TABLE)
      await db.execute(
        'DELETE FROM events WHERE uid = ? AND calendar_name = ?',
        (uid, self._calendar_name),
      )
      await db.commit()
    _log.info('delete uid=%s', uid)