feat: Basic read/write functionality for database

client/data/resources/database.dbml

@@ -16,6 +16,6 @@ Table posture {
     // Proportion of time the user is in the frame
     prop_in_frame REAL
 
-    period_start DATETIME
-    period_end DATETIME
+    period_start TIMESTAMP
+    period_end TIMESTAMP
 }

client/data/routines.py

@@ -1,44 +1,122 @@
-"""Data routines that can be integrated into main control flow."""
+"""Module for interacting with SQLite database"""
 
 import sqlite3
-from typing import Any
+
+from datetime import datetime
+from typing import Any, NamedTuple, Optional
 from importlib import resources
+
 from pydbml import PyDBML
 
-DATABASE_DEFINITION = resources.files("data.resources").joinpath("database.dbml")
-DATABASE_RESOURCE = resources.files("data.resources").joinpath("database.db")
+RESOURCES = resources.files("data.resources")
+DATABASE_DEFINITION = RESOURCES.joinpath("database.dbml")
+DATABASE_RESOURCE = RESOURCES.joinpath("database.db")
+
+
+class User(NamedTuple):
+    """Represents a user record in the SQLite database"""
+
+    id_: Optional[int]
+
+
+class Posture(NamedTuple):
+    """Represents a posture record in the SQLite database"""
+
+    id_: Optional[int]
+    user_id: int
+    prop_good: float
+    prop_in_frame: float
+    period_start: datetime
+    period_end: datetime
 
 
 def init_database() -> None:
     """Initialise SQLite database if it does not already exist"""
-    # Open connection with database
+    # Check if database exists
     with resources.as_file(DATABASE_RESOURCE) as database_file:
         if database_file.is_file():
             return
 
-        parsed = PyDBML(DATABASE_DEFINITION)
-        init_script = parsed.sql
-
-        connection = sqlite3.connect(database_file)
+    parsed = PyDBML(DATABASE_DEFINITION)
+    init_script = parsed.sql
 
     # Run init script
-    with connection:
+    with _connect() as connection:
         cursor = connection.cursor()
         cursor.executescript(init_script)
         connection.commit()
 
 
+def create_user() -> int:
+    """Creates a new user in the database.
+
+    Returns:
+        The id of the new user.
+    """
+    with _connect() as connection:
+        cursor = connection.cursor()
+        cursor.execute("INSERT INTO user DEFAULT VALUES;")
+        result = cursor.execute("SELECT last_insert_rowid() FROM user;")
+        user_id = result.fetchone()[0]
+        connection.commit()
+
+    return user_id
+
+
+def save_posture(posture: Posture | tuple) -> None:
+    """Stores the posture record in the database.
+
+    Args:
+        posture: The posture record to save.
+    """
+    if posture[0] is not None:
+        raise ValueError("Posture record id must be None")
+
+    with _connect() as connection:
+        cursor = connection.cursor()
+        cursor.execute(
+            "INSERT INTO posture VALUES (?, ?, ?, ?, ?, ?);",
+            posture,
+        )
+        connection.commit()
+
+
+def get_users(num: int = 10) -> list[User]:
+    """
+    Args:
+        num: Number of user to retrieve
+
+    Returns:
+        num users from the database.
+    """
+    with _connect() as connection:
+        cursor = connection.cursor()
+        result = cursor.execute("SELECT * FROM user LIMIT ?", (num,))
+        return [User(*record) for record in result.fetchall()]
+
+
+def get_postures(num: int = 10) -> list[Posture]:
+    """
+    Args:
+        num: Number of posture records to retrieve
+
+    Returns:
+        num posture records from the database.
+    """
+    with _connect() as connection:
+        cursor = connection.cursor()
+        result = cursor.execute("SELECT * FROM posture LIMIT ?", (num,))
+        return [Posture(*record) for record in result.fetchall()]
+
+
 def get_schema_info() -> list[list[tuple[Any]]]:
     """Column information on all tables in database.
 
     Returns:
-        (list[list[tuple[Any]]]): Outer list contains table information, inner list contains column
+        Outer list contains table information, inner list contains column
             information tuples.
     """
-    with resources.as_file(DATABASE_RESOURCE) as database_file:
-        connection = sqlite3.connect(database_file)
-
-    with connection:
+    with _connect() as connection:
         cursor = connection.cursor()
         result = cursor.execute("SELECT name FROM sqlite_schema WHERE type='table'")
         tables = result.fetchall()
@@ -50,3 +128,10 @@ def get_schema_info() -> list[list[tuple[Any]]]:
             table_schemas.append(table_schema)
 
     return table_schemas
+
+
+def _connect() -> sqlite3.Connection:
+    with resources.as_file(DATABASE_RESOURCE) as database_file:
+        return sqlite3.connect(
+            database_file, detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES
+        )

notebooks/test-database-tools.ipynb

@@ -2,7 +2,7 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 24,
+   "execution_count": 18,
    "metadata": {},
    "outputs": [
     {
@@ -21,16 +21,26 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 26,
+   "execution_count": 19,
    "metadata": {},
    "outputs": [],
    "source": [
-    "from data.routines import init_database, get_schema_info"
+    "import time\n",
+    "from datetime import datetime\n",
+    "from data.database import (\n",
+    "    init_database,\n",
+    "    get_schema_info,\n",
+    "    create_user,\n",
+    "    save_posture,\n",
+    "    get_users,\n",
+    "    get_postures,\n",
+    "    Posture,\n",
+    ")"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 27,
+   "execution_count": 20,
    "metadata": {},
    "outputs": [],
    "source": [
@@ -39,28 +49,113 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 29,
+   "execution_count": 21,
    "metadata": {},
    "outputs": [
     {
      "data": {
       "text/plain": [
-       "[[(0, 'user_id', 'INTEGER', 0, None, 0),\n",
-       "  (1, 'prop_good', 'REAL', 0, None, 0),\n",
-       "  (2, 'period_start', 'DATETIME', 0, None, 0),\n",
-       "  (3, 'period_end', 'DATETIME', 0, None, 0)],\n",
-       " [(0, 'id', 'INTEGER', 0, None, 1)],\n",
-       " [(0, 'name', '', 0, None, 0), (1, 'seq', '', 0, None, 0)]]"
+       "[[(0, 'id', 'INTEGER', 0, None, 1),\n",
+       "  (1, 'user_id', 'INTEGER', 0, None, 0),\n",
+       "  (2, 'prop_good', 'REAL', 0, None, 0),\n",
+       "  (3, 'prop_in_frame', 'REAL', 0, None, 0),\n",
+       "  (4, 'period_start', 'TIMESTAMP', 0, None, 0),\n",
+       "  (5, 'period_end', 'TIMESTAMP', 0, None, 0)],\n",
+       " [(0, 'name', '', 0, None, 0), (1, 'seq', '', 0, None, 0)],\n",
+       " [(0, 'id', 'INTEGER', 0, None, 1)]]"
       ]
      },
-     "execution_count": 29,
+     "execution_count": 21,
      "metadata": {},
      "output_type": "execute_result"
     }
    ],
    "source": [
     "get_schema_info()"
    ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 22,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "3"
+      ]
+     },
+     "execution_count": 22,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "create_user()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 23,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[User(id_=1), User(id_=2), User(id_=3)]"
+      ]
+     },
+     "execution_count": 23,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "get_users(num=10)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 24,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "time1 = datetime.now()\n",
+    "time.sleep(1)\n",
+    "time2 = datetime.now()\n",
+    "posture = Posture(None, 1, 0.8, 1.0, time1, time2)\n",
+    "save_posture(posture)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 25,
+   "metadata": {},
+   "outputs": [
+    {
+     "data": {
+      "text/plain": [
+       "[Posture(id_=1, user_id=1, prop_good=0.8, prop_in_frame=1.0, period_start=datetime.datetime(2024, 8, 18, 17, 40, 44, 458720), period_end=datetime.datetime(2024, 8, 18, 17, 40, 45, 462898)),\n",
+       " Posture(id_=2, user_id=1, prop_good=0.8, prop_in_frame=1.0, period_start=datetime.datetime(2024, 8, 18, 17, 41, 46, 12187), period_end=datetime.datetime(2024, 8, 18, 17, 41, 47, 13719)),\n",
+       " Posture(id_=3, user_id=1, prop_good=0.8, prop_in_frame=1.0, period_start=datetime.datetime(2024, 8, 18, 17, 46, 2, 389947), period_end=datetime.datetime(2024, 8, 18, 17, 46, 3, 402225)),\n",
+       " Posture(id_=4, user_id=1, prop_good=0.8, prop_in_frame=1.0, period_start=datetime.datetime(2024, 8, 18, 17, 48, 16, 460303), period_end=datetime.datetime(2024, 8, 18, 17, 48, 17, 466763))]"
+      ]
+     },
+     "execution_count": 25,
+     "metadata": {},
+     "output_type": "execute_result"
+    }
+   ],
+   "source": [
+    "get_postures(num=10)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
   }
  ],
  "metadata": {
@@ -79,7 +174,7 @@
    "name": "python",
    "nbconvert_exporter": "python",
    "pygments_lexer": "ipython3",
-   "version": "3.12.4"
+   "version": "3.10.14"
   }
  },
  "nbformat": 4,