wrapped.vim/python3/wrapped/db.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Database things"""

import datetime
import os
import sqlite3 as sql
import typing as t
from enum import Enum, auto

from . import util


class EventType(Enum):
    """Event types"""

    write = auto()  # Write event: File was written
    command = auto()  # Command event: Command was ran
    begin_s = auto()  # Session begin event: Begin a session
    end_s = auto()  # Session begin event: End a session
    copy = auto()  # Copy event: Copy to clipboard
    paste = auto()  # Paste event: Paste from clipboard


class Event:
    """Event class"""

    def __init__(
        self,
        id: int,
        type: str,
        utc_dt: str,
        file: str,
        language: str,
        session_id: str,
        window_id: int,
        buffer_id: int,
        data: t.Optional[str],
    ) -> None:
        self.id: int = id
        self.type: EventType = EventType[type]
        self.utc_dt: datetime.datetime = datetime.datetime.fromisoformat(utc_dt)
        self.file: str = file or "[No Name]"
        self.language: str = language or "text"
        self.session_id: str = session_id
        self.window_id: int = window_id
        self.buffer_id: int = buffer_id
        self.data: t.Optional[str] = data

    @classmethod
    def from_query(cls, q: t.Tuple[t.Any, ...]) -> "Event":
        """Return Event from SQL query result"""
        return cls(*q)


def vd() -> sql.Connection:
    """Vim database connection"""
    conn: sql.Connection = sql.connect(
        os.path.join(os.path.expanduser("~"), ".vim", "wrapped.db")
    )
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA synchronous=NORMAL")
    return conn


def init_stats_db() -> None:
    """Initialises statistics database"""

    valid: str = ", ".join(f"'{e.name}'" for e in EventType)

    with vd() as db:
        db.execute(
            f"""CREATE TABLE IF NOT EXISTS event (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    type TEXT NOT NULL CHECK(type IN ({valid})),
    utc_dt DATETIME NOT NULL,
    file TEXT NOT NULL,
    language TEXT NOT NULL,
    session_id TEXT NOT NULL,
    window_id INTEGER,
    buffer_id INTEGER,
    data TEXT
);"""
        )


def new_write_event(file: str, language: str, loc_add: int, loc_del: int) -> None:
    """Insert write event"""

    vid: util.VimID = util.get_vid()

    with vd() as db:
        db.execute(
            "INSERT INTO event (type, utc_dt, file, language, session_id, window_id, buffer_id, data) VALUES (?, ?, ?, ?, ?, ?, ?, ?);",
            (
                EventType.write.name,
                util.ts(),
                file,
                language,
                vid.session_id,
                vid.window_id,
                vid.buffer_id,
                f"{loc_add},{loc_del}",
            ),
        )


def new_command_event(command: str, file: str, language: str) -> None:
    """Insert command event"""

    vid: util.VimID = util.get_vid()

    with vd() as db:
        db.execute(
            "INSERT INTO event (type, utc_dt, file, language, session_id, window_id, buffer_id, data) VALUES (?, ?, ?, ?, ?, ?, ?, ?);",
            (
                EventType.command.name,
                util.ts(),
                file,
                language,
                vid.session_id,
                vid.window_id,
                vid.buffer_id,
                command,
            ),
        )


def new_session_event(
    begin: bool,
    file: str,
    language: str,
) -> None:
    """Insert new session event"""

    vid: util.VimID = util.get_vid()

    with vd() as db:
        db.execute(
            "INSERT INTO event (type, utc_dt, file, language, session_id, window_id, buffer_id) VALUES (?, ?, ?, ?, ?, ?, ?);",
            (
                EventType.begin_s.name if begin else EventType.end_s.name,
                util.ts(),
                file,
                language,
                vid.session_id,
                vid.window_id,
                vid.buffer_id,
            ),
        )


def new_clipboard_event(
    copying: bool,
    file: str,
    language: str,
) -> None:
    """Insert new clipboard event"""

    vid: util.VimID = util.get_vid()

    with vd() as db:
        db.execute(
            "INSERT INTO event (type, utc_dt, file, language, session_id, window_id, buffer_id) VALUES (?, ?, ?, ?, ?, ?, ?);",
            (
                EventType.copy.name if copying else EventType.paste.name,
                util.ts(),
                file,
                language,
                vid.session_id,
                vid.window_id,
                vid.buffer_id,
            ),
        )


def delete_all() -> None:
    """Deletes all statistics"""

    with vd() as db:
        db.execute("DELETE FROM event;")