initial commmit

pyproject.toml

@@ -0,0 +1,49 @@
+[tool.poetry]
+name = "grung-db"
+version = "1.0"
+description = "grung-db: A very small database toolkit."
+authors = ["evilchili"]
+readme = "README.md"
+packages = [
+    {include = "*", from = "src"},
+]
+
+[tool.poetry.dependencies]
+python = "^3.11"
+tinydb = "^4.8.2"
+nanoid = "^2.0.0"
+
+[tool.poetry.group.dev.dependencies]
+pytest = "*"
+pytest-cov = "*"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+
+### SLAM
+
+[tool.black]
+line-length = 120
+target-version = ['py311']
+
+[tool.isort]
+multi_line_output = 3
+line_length = 120
+include_trailing_comma = true
+
+[tool.autoflake]
+check = false                        # return error code if changes are needed
+in-place = true                      # make changes to files instead of printing diffs
+recursive = true                     # drill down directories recursively
+remove-all-unused-imports = true     # remove all unused imports (not just those from the standard library)
+ignore-init-module-imports = true    # exclude __init__.py when removing unused imports
+remove-duplicate-keys = true         # remove all duplicate keys in objects
+remove-unused-variables = true       # remove unused variables
+
+[tool.pytest.ini_options]
+log_cli_level = "DEBUG"
+addopts = "--cov=src --cov-report=term-missing"
+
+### ENDSLAM

src/grung/db.py

@@ -0,0 +1,82 @@
+import inspect
+
+from tinydb import TinyDB, table
+from tinydb.table import Document
+
+from grung.types import Record
+
+
+class RecordTable(table.Table):
+    """
+    Wrapper around tinydb Tables that handles Records instead of dicts.
+    """
+
+    def __init__(self, storage, name, document_class: Document = Record, **kwargs):
+        self.document_class = document_class
+        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 remove(self, document):
+        if document.doc_id:
+            super().remove(doc_ids=[document.doc_id])
+
+    def _satisfy_constraints(self, document):
+        # check for uniqueness, etc.
+        pass
+
+
+class GrungDB(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"
+    _tables = {}
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.create_table(Record)
+
+    def table(self, name: str) -> RecordTable:
+        if name not in self._tables:
+            raise RuntimeError(f"No such table: {name}")
+        return self._tables[name]
+
+    def create_table(self, table_class):
+        name = table_class.__name__
+        if name not in self._tables:
+            self._tables[name] = RecordTable(self.storage, name, document_class=table_class)
+        return self.table(name)
+
+    def save(self, record):
+        """
+        Create or update a record in its table.
+        """
+        return self.table(record._metadata.table).insert(record)
+
+    def delete(self, record):
+        return self.table(record._metadata.table).remove(record)
+
+    def __getattr__(self, attr_name):
+        """
+        Make tables attributes of the instance.
+        """
+        if attr_name in self._tables:
+            return self.table(attr_name)
+        return super().__getattr__(attr_name)
+
+    @classmethod
+    def with_schema(cls, schema_module, *args, **kwargs):
+        db = GrungDB(*args, **kwargs)
+        for name, obj in inspect.getmembers(schema_module):
+            if type(obj) == object and issubclass(Record, obj):
+                db.create_table(obj)
+        return db

src/grung/types.py

@@ -0,0 +1,47 @@
+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.
+    """
+
+    def __init__(self, raw_doc: dict = {}, doc_id: int = None, **params):
+        # populate the metadata
+        self._fields.append(
+            # 1% collision rate at ~2M records
+            Field("uid", default=nanoid.generate(size=8), unique=True)
+        )
+        self._metadata = Metadata(table=self.__class__.__name__, fields={f.name: f for f in self._fields})
+        self.doc_id = doc_id
+        super().__init__(dict({field.name: field.default for field in self._fields}, **raw_doc, **params))
+
+    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()}"

test/test_db.py

@@ -0,0 +1,64 @@
+import sys
+from typing import List
+
+import pytest
+from tinydb.storages import MemoryStorage
+
+from grung.db import GrungDB
+from grung.types import Field, Record
+
+
+class User(Record):
+    _fields = [Field("name"), Field("email", unique=True)]
+
+
+class Group(Record):
+    _fields = [Field("name", unique=True), Field("users", List[User])]
+
+
+@pytest.fixture
+def db():
+    _db = GrungDB.with_schema(sys.modules[__name__], storage=MemoryStorage)
+    _db.create_table(User)
+    _db.create_table(Group)
+    yield _db
+    print(_db)
+
+
+def test_crud(db):
+    user = User(name="john", email="john@foo")
+    assert user.uid
+    assert user._metadata.fields["uid"].unique
+
+    # insert
+    john_something = db.save(user)
+    last_insert_id = john_something.doc_id
+
+    # read back
+    assert 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 = db.User.get(doc_id=john_something.doc_id)
+    after_update = db.save(john_something)
+    assert after_update == john_something
+    assert before_update != after_update
+
+    # pointers
+    players = Group(name="players", users=[john_something])
+    players = db.save(players)
+    players.users[0]["name"] = "fnord"
+    db.save(players)
+
+    # modify records
+    players.users = []
+    db.save(players)
+    after_update = db.Group.get(doc_id=players.doc_id)
+    assert after_update.users == []
+
+    # delete
+    db.delete(players)
+    assert len(db.Group) == 0