Start using sqlite-migrate, add ended_timestamp column, refs #17

datasette_auth_tokens/__init__.py

@@ -2,6 +2,7 @@
 import itsdangerous
 import json
 import secrets
+import sqlite_utils
 import time
 from markupsafe import Markup
 from .views import (
@@ -11,20 +12,7 @@
     token_details,
     Config,
 )
-
-CREATE_TABLES_SQL = """
-CREATE TABLE _datasette_auth_tokens (
-    id INTEGER PRIMARY KEY,
-    token_status TEXT DEFAULT 'A', -- [A]ctive, [R]evoked, [E]xpired
-    description TEXT,
-    actor_id TEXT,
-    permissions TEXT,
-    created_timestamp INTEGER,
-    last_used_timestamp INTEGER,
-    expires_after_seconds INTEGER,
-    secret_version INTEGER DEFAULT 0
-);
-"""
+from .migrations import migration
 
 TOKEN_STATUSES = {
     "A": "Active",
@@ -64,18 +52,11 @@ def startup(datasette):
     db = config.db
 
     async def inner():
-        if "_datasette_auth_tokens" not in await db.table_names():
-            await db.execute_write(CREATE_TABLES_SQL)
-        else:
-            # Update any old 'L' tokens to 'A' - Live is now Active
-            # I'll remove this in a few versions
-            await db.execute_write(
-                """
-                update _datasette_auth_tokens
-                set token_status = 'A'
-                where token_status = 'L'
-                """
-            )
+        def migrate(conn):
+            db = sqlite_utils.Database(conn)
+            migration.apply(db)
+
+        await db.execute_write_fn(migrate)
 
     return inner
 

datasette_auth_tokens/migrations.py

@@ -0,0 +1,72 @@
+from sqlite_migrate import Migrations
+import time
+
+migration = Migrations("datasette_auth_tokens")
+
+
+# Use this decorator against functions that implement migrations
+@migration()
+def m001_create_table(db):
+    # If the table exists already, this will be a no-op
+    db.execute(
+        """
+    CREATE TABLE IF NOT EXISTS _datasette_auth_tokens (
+        id INTEGER PRIMARY KEY,
+        token_status TEXT DEFAULT 'A', -- [A]ctive, [R]evoked, [E]xpired
+        description TEXT,
+        actor_id TEXT,
+        permissions TEXT,
+        created_timestamp INTEGER,
+        last_used_timestamp INTEGER,
+        expires_after_seconds INTEGER,
+        secret_version INTEGER DEFAULT 0
+    );
+    """
+    )
+
+
+@migration()
+def m002_rename_live_to_active(db):
+    # In case anything is left over - I made this change before
+    # I introduced migrations
+    db["_datasette_auth_tokens"].transform(defaults={"token_status": "A"})
+    db.query(
+        """
+        update _datasette_auth_tokens
+        set token_status = 'A'
+        where token_status = 'L'
+        """
+    )
+
+
+@migration()
+def m003_add_ended_timestamp(db):
+    db["_datasette_auth_tokens"].add_column("ended_timestamp", int)
+    # Switch order around
+    db["_datasette_auth_tokens"].transform(
+        column_order=[
+            "id",
+            "token_status",
+            "description",
+            "actor_id",
+            "permissions",
+            "created_timestamp",
+            "last_used_timestamp",
+            "expires_after_seconds",
+            "ended_timestamp",
+            "secret_version",
+        ]
+    )
+    # Set it to now for any revoked tokens
+    db.query(
+        "update _datasette_auth_tokens set ended_timestamp = :now where token_status = 'R'",
+        {"now": int(time.time())},
+    )
+    # Set it to created_timestamp + expires_after_seconds for any expired tokens
+    db.query(
+        """
+        update _datasette_auth_tokens
+        set ended_timestamp = created_timestamp + expires_after_seconds
+        where token_status = 'E'
+        """
+    )

setup.py

@@ -31,6 +31,7 @@ def get_long_description():
     install_requires=[
         "datasette>=1.0a5",
         "sqlite-utils",
+        "sqlite-migrate",
     ],
     extras_require={"test": ["pytest", "pytest-asyncio", "httpx", "sqlite-utils"]},
     package_data={"datasette_auth_tokens": ["templates/*.html"]},

tests/test_migrations.py

@@ -0,0 +1,60 @@
+from datasette_auth_tokens.migrations import migration
+import sqlite_utils
+
+OLD_CREATE_TABLES_SQL = """
+CREATE TABLE _datasette_auth_tokens (
+    id INTEGER PRIMARY KEY,
+    token_status TEXT DEFAULT 'L', -- [L]ive, [R]evoked, [E]xpired
+    description TEXT,
+    actor_id TEXT,
+    permissions TEXT,
+    created_timestamp INTEGER,
+    last_used_timestamp INTEGER,
+    expires_after_seconds INTEGER,
+    secret_version INTEGER DEFAULT 0
+);
+"""
+
+
+def test_migrate_from_original():
+    db = sqlite_utils.Database(memory=True)
+    db.execute(OLD_CREATE_TABLES_SQL)
+    assert db["_datasette_auth_tokens"].columns_dict == {
+        "id": int,
+        "token_status": str,
+        "description": str,
+        "actor_id": str,
+        "permissions": str,
+        "created_timestamp": int,
+        "last_used_timestamp": int,
+        "expires_after_seconds": int,
+        "secret_version": int,
+    }
+
+    # Default token_status should be L
+    def get_col():
+        return [
+            col
+            for col in db["_datasette_auth_tokens"].columns
+            if col.name == "token_status"
+        ][0]
+
+    assert get_col().default_value == "'L'"
+    migration.apply(db)
+    assert db["_datasette_auth_tokens"].columns_dict["ended_timestamp"] == int
+    # Should have updated token default
+    assert get_col().default_value == "'A'"
+    # Confirm column order is correct
+    column_order = [col.name for col in db["_datasette_auth_tokens"].columns]
+    assert column_order == [
+        "id",
+        "token_status",
+        "description",
+        "actor_id",
+        "permissions",
+        "created_timestamp",
+        "last_used_timestamp",
+        "expires_after_seconds",
+        "ended_timestamp",
+        "secret_version",
+    ]