First draft of plugin, refs #1

.github/workflows/publish.yml

@@ -0,0 +1,50 @@
+name: Publish Python Package
+
+on:
+  release:
+    types: [created]
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+        cache: pip
+        cache-dependency-path: setup.py
+    - name: Install dependencies
+      run: |
+        pip install '.[test]'
+    - name: Run tests
+      run: |
+        pytest
+  deploy:
+    runs-on: ubuntu-latest
+    needs: [test]
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python
+      uses: actions/setup-python@v4
+      with:
+        python-version: "3.11"
+        cache: pip
+        cache-dependency-path: setup.py
+    - name: Install dependencies
+      run: |
+        pip install setuptools wheel twine build
+    - name: Publish
+      env:
+        TWINE_USERNAME: __token__
+        TWINE_PASSWORD: ${{ secrets.PYPI_TOKEN }}
+      run: |
+        python -m build
+        twine upload dist/*

.github/workflows/test.yml

@@ -0,0 +1,27 @@
+name: Test
+
+on: [push, pull_request]
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: ["3.8", "3.9", "3.10", "3.11"]
+    steps:
+    - uses: actions/checkout@v3
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+        cache: pip
+        cache-dependency-path: setup.py
+    - name: Install dependencies
+      run: |
+        pip install '.[test]'
+    - name: Run tests
+      run: |
+        pytest

pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "sqlite-migrate"
+version = "0.1a0"
+description = "A simple database migration system for SQLite, based on sqlite-utils"
+dependencies = [
+    "sqlite-utils"
+]
+
+[project.optional-dependencies]
+test = ["pytest"]
+
+[project.entry-points.sqlite_utils]
+migrate = "sqlite_migrate.sqlite_utils_plugin"

sqlite_migrate/__init__.py

@@ -0,0 +1,77 @@
+from dataclasses import dataclass
+import datetime
+from typing import Callable, Optional
+
+
+class Migrations:
+    migrations_table = "_sqlite_migrations"
+
+    @dataclass
+    class _Migration:
+        name: str
+        fn: Callable
+
+    def __init__(self, name: str):
+        """
+        :param name: The name of the migration set. This should be unique.
+        """
+        self.name = name
+        self._migrations = []
+
+    def __call__(self, *, name: Optional[str] = None) -> Callable:
+        """
+        :param name: The name to use for this migration - if not provided,
+          the name of the function will be used
+        """
+
+        def inner(func: Callable) -> Callable:
+            self._migrations.append(self._Migration(name or func.__name__, func))
+            return func
+
+        return inner
+
+    def pending(self, db: "sqlite_utils.Database"):
+        """
+        Return a list of pending migrations.
+        """
+        self.ensure_migrations_table(db)
+        already_applied = {r["name"] for r in db[self.migrations_table].rows}
+        return [
+            migration
+            for migration in self._migrations
+            if migration.name not in already_applied
+        ]
+
+    def apply(self, db: "sqlite_utils.Database"):
+        """
+        Apply any pending migrations to the database.
+        """
+        self.ensure_migrations_table(db)
+        for migration in self.pending(db):
+            migration.fn(db)
+            db[self.migrations_table].insert(
+                {
+                    "migration_set": self.name,
+                    "name": migration.name,
+                    "applied_at": str(datetime.datetime.utcnow()),
+                }
+            )
+
+    def ensure_migrations_table(self, db: "sqlite_utils.Database"):
+        """
+        Create _sqlite_migrations table if it doesn't already exist
+        """
+        if not db[self.migrations_table].exists():
+            db[self.migrations_table].create(
+                {
+                    "migration_set": str,
+                    "name": str,
+                    "applied_at": str,
+                },
+                pk="name",
+            )
+
+    def __repr__(self):
+        return "<Migrations '{}': [{}]>".format(
+            self.name, ", ".join(m.name for m in self._migrations)
+        )

sqlite_migrate/sqlite_utils_plugin.py

@@ -0,0 +1,59 @@
+import click
+import pathlib
+import sqlite_utils
+from sqlite_migrate import Migrations
+
+
+@sqlite_utils.hookimpl
+def register_commands(cli):
+    @cli.command()
+    @click.argument("path", type=click.Path(dir_okay=False))
+    @click.option(
+        "list_", "--list", is_flag=True, help="List migrations without running them"
+    )
+    def migrate(path, list_):
+        """
+        Apply pending database migrations.
+
+        Usage:
+
+            sqlite-utils migrate path/to/database.db
+
+        This will find the migrations.py file in the current directory
+        or subdirectories and apply any pending migrations.
+
+        Pass --list to see which migrations would be applied without
+        actually applying them.
+
+        Pass -m path/to/migrations.py to use a specific migrations file:
+
+            sqlite-utils migrate database.db -m path/to/migrations.py
+        """
+        # Find the migrations.py file
+        files = pathlib.Path(".").rglob("migrations.py")
+        migration_sets = []
+        for filepath in files:
+            code = filepath.read_text()
+            namespace = {}
+            exec(code, namespace)
+            # Find all instances of Migrations
+            for obj in namespace.values():
+                if isinstance(obj, Migrations):
+                    migration_sets.append(obj)
+        if not migration_sets:
+            raise click.ClickException(
+                "No migrations.py file found in current or subdirectories"
+            )
+        db = sqlite_utils.Database(path)
+        for migration_set in migration_sets:
+            if list_:
+                click.echo(
+                    "Pending migrations for {}:\n{}".format(
+                        path,
+                        "\n".join(
+                            "- {}".format(m.name) for m in migration_set.pending(db)
+                        ),
+                    )
+                )
+            else:
+                migration_set.apply(db)

tests/test_sqlite_migrate.py

@@ -0,0 +1,19 @@
+from sqlite_migrate import Migrations
+import sqlite_utils
+
+
+def test_basic():
+    db = sqlite_utils.Database(memory=True)
+    assert db.table_names() == []
+    migrations = Migrations("test")
+
+    @migrations()
+    def m001(db):
+        db["dogs"].insert({"name": "Cleo"})
+
+    @migrations()
+    def m002(db):
+        db["cats"].create({"name": str})
+
+    migrations.apply(db)
+    assert set(db.table_names()) == {"_sqlite_migrations", "dogs", "cats"}

tests/test_sqlite_utils_migrate_command.py

@@ -0,0 +1,41 @@
+from sqlite_migrate import Migrations
+import sqlite_utils
+from sqlite_utils.cli import cli
+import click
+import pathlib
+from click.testing import CliRunner
+
+
+def test_basic():
+    runner = CliRunner()
+    with runner.isolated_filesystem():
+        path = pathlib.Path(".")
+        (path / "foo").mkdir()
+        migrations_py = path / "foo" / "migrations.py"
+        migrations_py.write_text(
+            """
+from sqlite_migrate import Migrations
+
+m = Migrations("hello")
+
+@m()
+def foo(db):
+    db["foo"].insert({"hello": "world"})
+
+@m()
+def bar(db):
+    db["bar"].insert({"hello": "world"})
+    """,
+            "utf-8",
+        )
+        db_path = str(path / "test.db")
+        result = runner.invoke(sqlite_utils.cli.cli, ["migrate", db_path])
+        assert result.exit_code == 0, result.output
+        db = sqlite_utils.Database(db_path)
+        assert db["foo"].exists()
+        assert db["bar"].exists()
+        assert db["_sqlite_migrations"].exists()
+        rows = list(db["_sqlite_migrations"].rows)
+        assert len(rows) == 2
+        assert rows[0]["name"] == "foo"
+        assert rows[1]["name"] == "bar"