WIP

pyproject.toml

@@ -14,8 +14,9 @@ python-dotenv = "*"
 rich = "*"
 typer = "*"
 flask = "*"
-flask-sqlalchemy = "^3.1.1"
+tinydb = "^4.8.2"
-flask-migrate = "^4.1.0"
+pyyaml = "^6.0.2"
+nanoid = "^2.0.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "*"

src/ttfrog/app.py

@@ -1,50 +1,23 @@
-import importlib
 import os
+import sys
 
 from flask import Flask
-from flask_migrate import Migrate
+from tinydb.storages import MemoryStorage
-from flask_sqlalchemy import SQLAlchemy
 
-_context = None
+from ttfrog.db import Database
 
 
 class ApplicationContext:
-    """
+    def __init__(self):
-    Interface for the flask application
+        self.web: Flask = Flask("ttfrog")
-    """
+        self.db: Database = Database(storage=MemoryStorage)
-
-    def __init__(self, schema_module_name: str):
-        self.schema_module = importlib.import_module(schema_module_name)
-        self.db = SQLAlchemy()
-        self.migrate = Migrate()
-        self._app = Flask("flask")
-
         self._initialized = False
-        self._enable_migrations = False
 
-        self._app.config["SECRET_KEY"] = os.getenv("SECRET_KEY", "secret string")
+    def initialize(self, db: Database = None):
-        self._app.config["SQLALCHEMY_DATABASE_URI"] = os.getenv(
-            "DATABASE_URL", "sqlite:////" + os.path.join(self._app.root_path, "data.db")
-        )
-        self._app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
-
-    @property
-    def flask(self):
         if not self._initialized:
-            raise RuntimeError("You must initialize first.")
+            self.web.config["SECRET_KEY"] = os.getenv("SECRET_KEY", "secret string")
-        return self._app
+            self.db = db or Database("ttfrog.db.json")
-
-    def initialize(self):
-        self.db.init_app(self._app)
-        if self._enable_migrations:
-            self.migrate.init_app(self._app, self.db)
             self._initialized = True
-        return self.flask
 
 
-def initialize(schema_module_name: str = "ttfrog.schema"):
+sys.modules[__name__] = ApplicationContext()
-    global _context
-    if not _context:
-        _context = ApplicationContext(schema_module_name)
-        _context.initialize()
-    return _context

src/ttfrog/cli.py

@@ -20,7 +20,7 @@ LOG_LEVEL=INFO
 main_app = typer.Typer()
 
 _context = ttfrog.app.initialize()
-flask_app = _context.flask
+flask_app = _context.app
 
 app_state = dict(
     config_file=Path("~/.config/ttfrog.conf").expanduser(),
@@ -67,19 +67,7 @@ def run(context: typer.Context):
 
 @flask_app.shell_context_processor
 def make_shell_context():
-    return {"db": flask_app.db, "app": flask_app._app}
+    return flask_app
-
-
-@flask_app.cli.command()
-@click.option("--drop", is_flag=True, help="Create after drop.")
-@click.pass_context
-def setup(ctx, drop: bool):
-    """
-    (Re)create the database.
-    """
-    if drop:
-        ctx.db.drop_all()
-    ctx.db.create_all()
 
 
 @click.group(cls=FlaskGroup, create_app=lambda: flask_app)

src/ttfrog/db.py

@@ -0,0 +1,53 @@
+from tinydb import TinyDB, table
+from tinydb.table import Document
+
+from ttfrog import schema
+
+
+class RecordTable(table.Table):
+    """
+    Wrapper around tinydb Tables that handles Records instead of dicts.
+    """
+
+    def __init__(self, storage, name, **kwargs):
+        self.document_class = getattr(schema, name, Document)
+        super().__init__(storage, name, **kwargs)
+
+    def insert(self, document):
+        self._satisfy_constraints(document)
+        if document.doc_id:
+            last_insert_id = super().upsert(document)[0]
+        else:
+            last_insert_id = super().insert(dict(document))
+        return self.get(doc_id=last_insert_id)
+
+    def _satisfy_constraints(self, document):
+        # check for uniqueness, etc.
+        pass
+
+
+class Database(TinyDB):
+    """
+    A TinyDB database instance that uses RecordTable instances for each table
+    and Record instances for each document in the table.
+    """
+
+    default_table_name = "Record"
+
+    def table(self, name: str, **kwargs) -> RecordTable:
+        if name not in self._tables:
+            self._tables[name] = RecordTable(self.storage, name, **kwargs)
+        return self._tables[name]
+
+    def save(self, record):
+        """
+        Create or update a record in its table.
+        """
+        return self.table(record._metadata.table).insert(record)
+
+    def __getattr__(self, attr_name):
+        """
+        Make tables attributes of the instance.
+        """
+        if attr_name in self.tables():
+            return self.table(attr_name)

src/ttfrog/schema.ph

@@ -1 +0,0 @@
-# schema goes here

src/ttfrog/schema.py

@@ -0,0 +1,58 @@
+from collections import namedtuple
+from dataclasses import dataclass
+
+import nanoid
+
+Metadata = namedtuple("Metadata", ["table", "fields"])
+
+
+@dataclass
+class Field:
+    """
+    Represents a single field in a Record.
+    """
+
+    name: str
+    value_type: type = str
+    default: value_type | None = None
+    unique: bool = False
+
+
+class Record(dict):
+    """
+    Base type for a single database record.
+    """
+
+    _fields = [Field("uid", default="", unique=True)]
+
+    def __init__(self, raw_doc: dict = {}, doc_id: int = None, **params):
+        # populate the metadata
+        fields = Record._fields
+        if self.__class__ != Record:
+            fields += self._fields
+        self._metadata = Metadata(table=self.__class__.__name__, fields={f.name: f for f in fields})
+
+        self.doc_id = doc_id
+
+        vals = dict({field.name: field.default for field in fields}, **raw_doc, **params)
+        if not vals["uid"]:
+            vals["uid"] = nanoid.generate(size=8)  # 1% collision rate at ~2M records
+
+        super().__init__(vals)
+
+    def __setattr__(self, key, value):
+        if key in self:
+            self[key] = value
+        super().__setattr__(key, value)
+
+    def __getattr__(self, attr_name):
+        if attr_name in self:
+            return self.get(attr_name)
+        return super().__getattr__(attr_name)
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}[{self.doc_id}]: {self.items()}"
+
+
+class User(Record):
+    _fields = [Field("name"), Field("email", unique=True)]

test/test_db.py

@@ -0,0 +1,39 @@
+import os
+
+import pytest
+
+import ttfrog.app
+from ttfrog import schema
+from ttfrog.db import Database
+
+
+@pytest.fixture
+def app():
+    ttfrog.app.initialize(Database("ttfrog-tests"))
+    ttfrog.app.db.drop_tables()
+    yield ttfrog.app
+    ttfrog.app.db.close()
+    os.unlink("ttfrog-tests")
+
+
+def test_create(app):
+    user = schema.User(name="john", email="john@foo")
+    assert user.uid
+    assert user._metadata.fields["uid"].unique
+
+    # insert
+    john_something = app.db.save(user)
+    last_insert_id = john_something.doc_id
+
+    # read back
+    assert app.db.User.get(doc_id=last_insert_id) == john_something
+    assert john_something.name == user.name
+    assert john_something.email == user.email
+    assert john_something.uid == user.uid
+
+    # update
+    john_something.name = "james?"
+    before_update = app.db.User.get(doc_id=john_something.doc_id)
+    after_update = app.db.save(john_something)
+    assert after_update == john_something
+    assert before_update != after_update

test/test_ttfrog_cli.py

@@ -1,8 +0,0 @@
-import pytest
-
-from ttfrog import cli
-
-
-@pytest.mark.xfail
-def test_tests_are_implemented():
-    assert cli.main()