reactivedb.py

from typing import TypeVar, Type, Optional, ClassVar, Any, Dict, List, get_type_hints
from dataclasses import dataclass, fields, is_dataclass, field
import inspect
import sqlite3
import json, asyncio
import pathlib
from pydantic import BaseModel, Field
from typing import get_origin, get_args, Union
from functools import partial

import sqlite3, reactive, threading, weakref
from typing import Protocol, Any, Callable, Mapping, List, Tuple, Dict, Optional, Iterator, TypeVar, Generic
from dataclasses import dataclass
from pyrsistent import pmap
from pyrsistent.typing import PMap
from frozendict import frozendict
from abc import ABC, abstractmethod
import psycopg2
import psycopg2.extras
import psycopg2.extensions
import queue, time, traceback

T = TypeVar('T', bound='Model')
R = TypeVar('R')
Id = int
Row = frozendict[str, Any]
PossiblyMutableRow = Mapping[str, Any]

class _Reducer(Protocol):
    def add(self, row: Row) -> None: ...
    def remove(self, row: Row) -> None: ...
    def set(self, rows: List[Row]) -> None: ...
    def value(self) -> Any: ...

_ReducerFactory = Callable[[], _Reducer]

class DictReducer(_Reducer):
    def __init__(self):
        self._value: PMap[Id, Row] = pmap()

    def set(self, rows: List[Row]) -> None:
        self._value = pmap({ row['id']: row for row in rows })

    def remove(self, row: Row) -> None:
        id = row['id']
        self._value = self._value.remove(id) if id in self._value else self._value

    def add(self, row: Row) -> None:
        self._value = self._value.set(row['id'], row)

    def value(self) -> Mapping[Any, Row]:
        return self._value

@dataclass(frozen=True)
class PartialWithEquality:
    f: Callable[..., Any]
    args: Tuple[Any, ...]

    def __call__(self) -> Any:
        return self.f(*self.args)

class AggReducer(_Reducer):
    @staticmethod
    def create(
        key: Callable[[Row], Any],
        add_f: Callable[[R|None, Row], R],
        remove_f: Callable[[R|None, Row], R]
    ) -> PartialWithEquality:
        return PartialWithEquality(AggReducer, args=(key, add_f, remove_f))
    
    def __init__(
        self,
        key: Callable[[Row], Any],
        add_f: Callable[[Any, Row], Any],
        remove_f: Callable[[Any, Row], Any]
    ):
        self._value: Dict[Any, Any] = {}
        self._key = key
        self._add_f = add_f
        self._remove_f = remove_f

    def set(self, rows: List[Row]) -> None:
        self._value = {}
        for row in rows: self.add(row)

    def add(self, row: Row) -> None:
        key = self._key(row)
        self._value[key] = self._add_f(self._value.get(key), row)
    
    def remove(self, row: Row) -> None:
        key = self._key(row)
        self._value[key] = self._remove_f(self._value.get(key), row)
    
    def value(self) -> Dict[Any, Any]:
        return self._value


class CountReducer(AggReducer):
    @staticmethod
    def create(key: Callable[[Row], Any]) -> PartialWithEquality: # type: ignore
        return AggReducer.create(
            key,
            add_f=lambda x, row: (x or 0) + 1,
            remove_f=lambda x, row: (x or 0) - 1
        )
    
class _Watch:
    def __init__(
        self,
        table: '_ReactiveTable',
        query: Mapping[str, Any],
        reducer: _ReducerFactory
    ):
        self.table: _ReactiveTable = table
        self.query: Mapping[str, Any] = query
        self.mask: Tuple[str, ...] = tuple(self.query.keys())
        self.masked_val: Tuple[Any, ...] = tuple(self.query.values())
        self.enabled: bool = False
        result = table._query_impl(query)
        self.reducer: _Reducer = reducer()
        self.reducer.set(result)
        self.var: reactive.CustomRef = reactive.CustomRef(
            self.reducer.value(),
            write_callback=None,
            enable_callback=self._enable,
            disable_callback=self._disable,
            _allow_in_immutable_ctx=True,
        )

    def delete_row(self, row: Row) -> None:
        self.reducer.remove(row)
        self.var.change_value(self.reducer.value())

    def insert_row(self, row: Row) -> None:
        self.reducer.add(row)
        self.var.change_value(self.reducer.value())

    def _enable(self) -> None:
        result = self.table._query_impl(self.query)
        self.reducer.set(result)
        self.var.change_value(self.reducer.value())
        self.enabled = True
        
    def _disable(self) -> None:
        self.enabled = False

class _ReactiveTable(ABC):
    def __init__(self):
        self._by_mask: Dict[
            Tuple[str, ...],
            Dict[
                Tuple[Any, ...],
                weakref.WeakValueDictionary[_ReducerFactory, _Watch]
            ]
        ] = {}
        self._lock: threading.RLock = threading.RLock()

    @abstractmethod
    def impl_get(self, id: Any) -> Optional[Row]:
        pass

    @abstractmethod
    def impl_delete(self, id: Any) -> None:
        pass

    @abstractmethod
    def impl_update(self, row: Row) -> None:
        pass

    @abstractmethod
    def impl_insert(self, row: Row) -> Any:
        pass

    @abstractmethod
    def _query_impl(self, query: Mapping[str, Any]) -> List[Row]:
        pass

    def _get_watch(
        self,
        query: Mapping[str, Any],
        reducer: _ReducerFactory
    ) -> _Watch:
        mask = tuple(query.keys())
        masked_values = tuple(query.values())
        mask_dict = self._by_mask.setdefault(mask, {})
        masked_values_dict = mask_dict.setdefault(
            masked_values,
            weakref.WeakValueDictionary()
        )
        r = masked_values_dict.get(reducer)
        if r is None:
            r = _Watch(self, query, reducer)
            masked_values_dict[reducer] = r
        return r

    def query(
        self,
        query: Mapping[str, Any],
        reducer: _ReducerFactory = DictReducer
    ) -> Any:
        r = reducer()
        result = self._query_impl(query)
        r.set(result)
        return r.value()

    def query_reactive(
        self,
        query: Mapping[str, Any],
        reducer: _ReducerFactory = DictReducer
    ) -> reactive.CustomRef:
        with self._lock:
            return self._get_watch(query, reducer).var
        
    def _watches_for_row(self, row: frozendict[str, Any]) -> Iterator[_Watch]:
        for mask, mask_dict in self._by_mask.items():
            masked = tuple(row.get(name) for name in mask)
            masked_values_dict = mask_dict.get(masked)
            if masked_values_dict:
                for watch in masked_values_dict.values():
                    if watch.enabled:
                        yield watch
    
    def _deleted_existing(self, row: Row) -> None:
        row_frozen = frozendict(row)
        for w in self._watches_for_row(row_frozen):
            w.delete_row(row_frozen)

    def _inserted(self, row: Row) -> None:
        row_frozen = frozendict(row)
        for w in self._watches_for_row(row_frozen):
            w.insert_row(row_frozen)

class _ReactiveWritesTable(_ReactiveTable):
    def delete(self, row: PossiblyMutableRow) -> None:
        reactive.assert_mutable_context()
        with self._lock:
            old_row = self.impl_get(row['id'])
            if old_row is not None:
                self.impl_delete(row['id'])
                self._deleted_existing(old_row)

    def set(self, row: PossiblyMutableRow) -> None:
        reactive.assert_mutable_context()
        with self._lock:
            frozen_row = frozendict(row)
            old_row = self.impl_get(frozen_row['id'])
            self.impl_update(frozendict(row))
            if old_row is not None:
                self._deleted_existing(old_row)
            self._inserted(frozen_row)

    def insert(self, row: PossiblyMutableRow) -> int:
        reactive.assert_mutable_context()
        frozen_row = frozendict(row)
        assert frozen_row.get('id') is None
        with self._lock:
            id = self.impl_insert(frozen_row)
            frozen_row = frozendict({**frozen_row, 'id': id})
            self._inserted(frozen_row)
        return id

class _ReactiveReplicationTable(_ReactiveTable):
    def delete(self, row: PossiblyMutableRow) -> None:
        reactive.assert_mutable_context()
        with self._lock:
            if self.impl_get(row['id']) is not None:
                self.impl_delete(row['id'])

    def set(self, row: PossiblyMutableRow) -> None:
        reactive.assert_mutable_context()
        with self._lock:
            frozen_row = frozendict(row)
            self.impl_update(frozen_row)

    def insert(self, row: PossiblyMutableRow) -> int:
        reactive.assert_mutable_context()
        frozen_row = frozendict(row)
        assert frozen_row.get('id') is None
        with self._lock:
            return self.impl_insert(frozen_row)

class SqliteTable(_ReactiveWritesTable):
     
    def __init__(
        self,
        db: Callable[[], sqlite3.Connection],
        table_name: str,
        columns: dict[str, str],
        index_columns: list[list[str]]
    ):
        super().__init__()
        self.db: Callable[[], sqlite3.Connection] = db
        self.table_name: str = table_name
        self.columns: dict[str, str] = columns
        self.index_columns: list[list[str]] = index_columns
        self._init()
        
    def _init(self):
        db = self.db()
        columns_sql = ', '.join(f"{name} {type_}" for name, type_ in self.columns.items())
        create_table_sql = f"""
            CREATE TABLE IF NOT EXISTS {self.table_name} (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                {columns_sql}
            )
        """
        db.execute(create_table_sql)
        
        existing_columns_query = f"PRAGMA table_info({self.table_name})"
        existing_columns_result = db.execute(existing_columns_query).fetchall()
        existing_columns = set()
        for row in existing_columns_result:
            existing_columns.add(row[1])  # 'name' column is at index 1

        columns_to_add = set(self.columns.keys()) - existing_columns

        for column_name in columns_to_add:
            column_type = self.columns[column_name]
            alter_table_sql = f"ALTER TABLE {self.table_name} ADD COLUMN {column_name} {column_type}"
            db.execute(alter_table_sql)
    
        existing_indexes: set[str] = set()
        for row in db.execute(
            f"SELECT name FROM sqlite_master WHERE type='index' AND tbl_name=?",
            (self.table_name,)
        ):
            existing_indexes.add(row[0])
        
        for idx_columns in self.index_columns:
            idx_name = f"{self.table_name}_{'_'.join(idx_columns)}_idx"
            if idx_name not in existing_indexes:
                cols = ', '.join(idx_columns)
                db.execute(f"CREATE INDEX {idx_name} ON {self.table_name}({cols})")
            existing_indexes.discard(idx_name)
        
        for idx_name in existing_indexes:
            db.execute(f"DROP INDEX {idx_name}")
    
    def impl_delete(self, id: Any) -> None:
        self.db().execute(f"DELETE FROM {self.table_name} WHERE id = ?", (id,))
    
    def impl_get(self, id: Any) -> Optional[Row]:
        cursor = self.db().execute(f"SELECT * FROM {self.table_name} WHERE id = ?", (id,))
        row = cursor.fetchone()
        if row:
            columns = [desc[0] for desc in cursor.description]
            return frozendict(zip(columns, row))
        return None
    
    def impl_update(self, row: Row) -> None:
        cols = list(row.keys())
        placeholders = ', '.join('?' * len(cols))
        col_names = ', '.join(cols)
        update_cols = [col for col in cols if col != 'id']
        updates = ', '.join(f"{col} = excluded.{col}" for col in update_cols)
        values = [row[col] for col in cols]

        self.db().execute(
            f"""
            INSERT INTO {self.table_name} ({col_names})
            VALUES ({placeholders})
            ON CONFLICT(id) DO UPDATE SET {updates}
            """,
            values
        )
    
    def impl_insert(self, row: Row) -> Any:
        cols = list(row.keys())
        placeholders = ', '.join('?' * len(cols))
        col_names = ', '.join(cols)
        values = [row[col] for col in cols]
        
        cursor = self.db().execute(
            f"INSERT INTO {self.table_name} ({col_names}) VALUES ({placeholders})",
            values
        )
        return cursor.lastrowid

    def _query_impl(self, query: Mapping[str, Any]) -> List[Row]:
        if not query:
            cursor = self.db().execute(f"SELECT * FROM {self.table_name}")
        else:
            where_clauses = [f"{col} = ?" for col in query.keys()]
            where_sql = " AND ".join(where_clauses)
            values = list(query.values())
            cursor = self.db().execute(
                f"SELECT * FROM {self.table_name} WHERE {where_sql}",
                values
            )
        columns = [desc[0] for desc in cursor.description]
        return [frozendict(zip(columns, row)) for row in cursor]

# Postgres is a multi-user database, so we'll need to call functions like _inserted/_deleted_existing based on logical replication stream, instead of the operations users do
class PostgresTable(_ReactiveReplicationTable):        
    def __init__(
        self,
        dsn: str,
        table_name: str,
        columns: dict[str, str],
        index_columns: list[list[str]],
        replication_slot: str,
        publication: str
    ):
        super().__init__()
        self._main_event_loop = asyncio.get_event_loop()
        self.dsn = dsn
        self.table_name = table_name
        self.columns = columns
        self.index_columns = index_columns
        self.replication_slot = replication_slot
        self.publication = publication
        self._init()
        self._changes_queue: queue.Queue[Any] = queue.Queue()
        self._closed = False
        
        self._conn_local = threading.local()  # Thread-local connection

        # Setup replication
        with psycopg2.connect(self.dsn) as conn:
            with conn.cursor() as cur:
                cur.execute(f'ALTER TABLE {self.table_name} REPLICA IDENTITY FULL;')

                # Create publication if it doesn't exist
                cur.execute("SELECT 1 FROM pg_publication WHERE pubname = %s", (self.publication,))
                if not cur.fetchone():
                    cur.execute(f"CREATE PUBLICATION {self.publication} FOR TABLE {self.table_name}")
                # Add table to publication if not already there
                #cur.execute(f"ALTER PUBLICATION {self.publication} ADD TABLE {self.table_name}")
            conn.commit()

        self._replication_thread_ready = threading.Event()
        self._replication_thread = threading.Thread(
            target=self._listen_replication,
            daemon=True
        )
        self._replication_thread.start()
        self._replication_thread_ready.wait(timeout=10)

    def _init(self):
        conn = psycopg2.connect(self.dsn)
        cursor = conn.cursor()
        cols = ', '.join(f"{name} {type_}" for name, type_ in self.columns.items())
        cursor.execute(
            f"CREATE TABLE IF NOT EXISTS {self.table_name} ("
            f"id SERIAL PRIMARY KEY, {cols})"
        )
        conn.commit()
        existing_cols_query = (
            "SELECT column_name FROM information_schema.columns "
            "WHERE table_name = %s"
        )
        cursor.execute(existing_cols_query, (self.table_name,))
        existing_cols = set(r[0] for r in cursor.fetchall())
        for col_name, col_type in self.columns.items():
            if col_name not in existing_cols:
                cursor.execute(
                    f"ALTER TABLE {self.table_name} ADD COLUMN {col_name} {col_type}"
                )
        conn.commit()
        idx_query = (
            "SELECT indexname FROM pg_indexes WHERE tablename = %s"
        )
        cursor.execute(idx_query, (self.table_name,))
        existing_indexes = set(r[0] for r in cursor.fetchall())
        for icols in self.index_columns:
            idx_name = f"{self.table_name}_{'_'.join(icols)}_idx"
            if idx_name not in existing_indexes:
                cols_joined = ', '.join(icols)
                cursor.execute(
                    f"CREATE INDEX {idx_name} ON {self.table_name}({cols_joined})"
                )
        conn.commit()
        cursor.close()
        conn.close()

    def _conn(self):
        if not hasattr(self._conn_local, 'conn'):
            self._conn_local.conn = psycopg2.connect(self.dsn)
        return self._conn_local.conn

    def commit(self):
        self._conn().commit()

    def impl_delete(self, id: Any) -> None:
        conn = self._conn()
        cursor = conn.cursor()
        cursor.execute(f"DELETE FROM {self.table_name} WHERE id = %s", (id,))
        cursor.close()
        conn.commit()  # ensure WAL is advanced

    def impl_get(self, id: Any) -> Optional[Row]:
        conn = self._conn()
        cursor = conn.cursor()
        cursor.execute(f"SELECT * FROM {self.table_name} WHERE id = %s", (id,))
        row = cursor.fetchone()
        if row is None:
            return None
        assert cursor.description is not None
        cols = [desc[0] for desc in cursor.description]
        return frozendict(zip(cols, row))

    def impl_update(self, row: Row) -> None:
        conn = self._conn()
        cursor = conn.cursor()
        cols = list(row.keys())
        placeholders = ', '.join('%s' for _ in cols)
        col_names = ', '.join(cols)
        update_cols = [col for col in cols if col != "id"]
        updates = ', '.join(f"{col} = excluded.{col}" for col in update_cols)
        values = [row[col] for col in cols]

        cursor.execute(
            f"""
            INSERT INTO {self.table_name} ({col_names})
            VALUES ({placeholders})
            ON CONFLICT(id) DO UPDATE SET {updates}
            """,
            values
        )
        cursor.close()
        conn.commit()  # ensure WAL is advanced

    def impl_insert(self, row: Row) -> Any:
        conn = self._conn()
        cursor = conn.cursor()
        cols = [c for c in row.keys() if c != 'id']
        placeholders = ", ".join(["%s"] * len(cols))
        col_names = ", ".join(cols)
        values = [row[c] for c in cols]

        cursor.execute(
            f"INSERT INTO {self.table_name} ({col_names}) "
            f"VALUES ({placeholders}) RETURNING id",
            values
        )
        resp = cursor.fetchone()
        assert resp, resp
        cursor.close()
        conn.commit()  # ensure WAL is advanced
        return resp[0]

    def _query_impl(self, query: Mapping[str, Any]) -> List[Row]:
        conn = self._conn()
        cursor = conn.cursor()
        if not query:
            cursor.execute(f"SELECT * FROM {self.table_name}")
        else:
            where_clauses = [f"{col} = %s" for col in query.keys()]
            sql = (f"SELECT * FROM {self.table_name} WHERE "
                   + " AND ".join(where_clauses))
            cursor.execute(sql, tuple(query.values()))
        desc = cursor.description
        if not desc:
            return []
        columns = [col[0] for col in desc]
        return [frozendict(zip(columns, row)) for row in cursor.fetchall()]

    def _listen_replication(self):
        while True:
            try:
                conn = psycopg2.connect(
                    self.dsn,
                    connection_factory=psycopg2.extras.LogicalReplicationConnection
                )
                cur = conn.cursor()
                try:
                    cur.execute(f"DROP_REPLICATION_SLOT {self.replication_slot}")
                except psycopg2.Error:
                    pass
                cur.create_replication_slot(self.replication_slot, output_plugin='wal2json')
                cur.start_replication(
                    slot_name=self.replication_slot,
                    options={
                        'format-version': '1'
                    },
                    decode=True
                )
                self._replication_thread_ready.set()
                
                cur.consume_stream(self._handle_replication_message)
            except psycopg2.Error as e:
                if self._closed or self._main_event_loop.is_closed(): break
                print(f"Replication error: {e}")
                time.sleep(1)  # Avoid tight loop on persistent errors
            finally:
                conn.close()

    def close(self):
        self._closed = True

    def _handle_replication_message(self, msg):
        try:
            data = msg.payload
            parsed = json.loads(data)
            changes = parsed.get('change', [])

            relevant_changes = []
            for change in changes:
                if change['table'] == self.table_name:
                    relevant_changes.append(change)

            if relevant_changes:
                self._main_event_loop.call_soon_threadsafe(
                    partial(self._apply_transaction_changes, relevant_changes)
                )

            msg.cursor.send_feedback(flush_lsn=msg.data_start)
        except Exception as e:
            traceback.print_exc()
            print(f"Error handling replication message: {e}")

    def _apply_transaction_changes(self, transaction_changes):
        if self._main_event_loop.is_closed():
            return
        with self._lock:
            for change in transaction_changes:
                if change['kind'] == 'insert':
                    row_frozen = frozendict(zip(change['columnnames'], change['columnvalues']))
                    self._inserted(row_frozen)
                elif change['kind'] == 'update':
                    old_row = frozendict(zip(change['oldkeys']['keynames'], change['oldkeys']['keyvalues']))
                    new_row = frozendict(zip(change['columnnames'], change['columnvalues']))
                    self._deleted_existing(old_row)
                    self._inserted(new_row)
                elif change['kind'] == 'delete':
                    row_frozen = frozendict(zip(change['oldkeys']['keynames'], change['oldkeys']['keyvalues']))
                    self._deleted_existing(row_frozen)

class Model(BaseModel):
    id: Optional[int] = Field(default=None)

    class TableMeta:
        indexes: ClassVar[List[List[str]]] = []

    def __init_subclass__(cls):
        super().__init_subclass__()
        cls._columns = cls._get_columns() # type: ignore

    @classmethod
    def _get_columns(cls) -> Dict[str, str]:
        hints = get_type_hints(cls)
        type_map = {
            int: 'INTEGER',
            str: 'TEXT',
            float: 'REAL',
            bool: 'INTEGER',
            pathlib.Path: 'TEXT',
        }

        columns = {}
        for name, type_ in hints.items():
            if name == 'id':
                continue

            sqlite_type = None
            actual_type = type_

            origin = get_origin(type_)
            if origin is Union:
                args = get_args(type_)
                if len(args) == 2 and type(None) in args:
                    actual_type = args[0] if args[1] is type(None) else args[1]

            sqlite_type = type_map.get(actual_type, 'JSON')
            columns[name] = sqlite_type
        return columns

    def to_row(self) -> Row:
        data = self.model_dump(mode='json')
        for name, type_ in self._columns.items(): # type: ignore
            if type_ == 'JSON' and data[name] is not None:
                data[name] = json.dumps(data[name])
        return frozendict(data)

    @classmethod
    def from_row(cls: Type[T], row: Row) -> T:
        data = dict(row)
        for name, type_ in cls._columns.items(): # type: ignore
            if type_ == 'JSON' and data[name] is not None:
                data[name] = json.loads(data[name])

        return cls.model_validate(data)

class ModelDictReducer(_Reducer):
    @staticmethod
    def create(model_cls: Type[T]) -> PartialWithEquality:
        return PartialWithEquality(ModelDictReducer, args=(model_cls,))

    def __init__(self, model_cls: Type[Model]):
        self._value: PMap[Id, Model] = pmap()
        self.model_cls = model_cls

    def set(self, rows: List[Row]) -> None:
        self._value = pmap({ row['id']: self.model_cls.from_row(row) for row in rows })

    def remove(self, row: Row) -> None:
        id = row['id']
        self._value = self._value.remove(id) if id in self._value else self._value

    def add(self, row: Row) -> None:
        model_instance = self.model_cls.from_row(row)
        self._value = self._value.set(row['id'], model_instance)

    def value(self) -> PMap[Any, Model]: # type: ignore
        return self._value

class TypedSqliteTable(Generic[T]):
    def __init__(self, db: Callable[[], sqlite3.Connection], model_cls: Type[T]):
        self.model_cls = model_cls
        self._sql_table = SqliteTable(
            db=db,
            table_name=model_cls.__name__.lower(),
            columns=model_cls._get_columns(),
            index_columns=model_cls.TableMeta.indexes
        )

    def query_reactive(
        self,
        reducer: Optional[_ReducerFactory] = None,
        **kwargs: Any
    ) -> reactive.Ref[Dict[Id, T]]:
        if reducer is None:
            reducer = ModelDictReducer.create(self.model_cls)
        ref = self._sql_table.query_reactive(kwargs, reducer)
        return ref

    def query(self, reducer: Optional[_ReducerFactory] = None, **kwargs: Any):
        if reducer is None:
            reducer = ModelDictReducer.create(self.model_cls)
        return self._sql_table.query(kwargs, reducer)

    def get(self, **kwargs: Any):
        r = self.query(**kwargs)
        if len(r) != 1:
            raise ValueError('expected exactly one result got %d' % len(r))
        return list(r.values())[0]
        
    def insert(self, model: T) -> None:
        assert isinstance(model, self.model_cls), (type(model), self.model_cls)
        reactive.assert_mutable_context()
        if getattr(model, 'id', None) is not None:
            raise ValueError("Cannot insert model with existing id")
        row = model.to_row()
        id = self._sql_table.insert(row)
        model.id = id  # type: ignore

    def set(self, model: T, only_columns=None) -> None:
        assert isinstance(model, self.model_cls), (type(model), self.model_cls)
        reactive.assert_mutable_context()
        if getattr(model, 'id', None) is None:
            raise ValueError("Cannot update model without id")
        row = model.to_row()
        self._sql_table.set(frozendict({**row, 'id': model.id})) # type: ignore

    def delete(self, model: T) -> None:
        assert isinstance(model, self.model_cls), (type(model), self.model_cls)
        reactive.assert_mutable_context()
        if getattr(model, 'id', None) is None:
            raise ValueError("Cannot delete model without id")
        self._sql_table.delete({'id': model.id})

class Db:
    def __init__(self, url):
        self._url = url
        self._tables = {}
        self._lock = threading.Lock()
        self._conn_local = threading.local()

    def _conn(self):
        if not hasattr(self._conn_local, 'conn'):
            self._conn_local.conn = sqlite3.connect(self._url)
        return self._conn_local.conn
        
    def table(self, model):
        with self._lock:
            if model not in self._tables:
                self._tables[model] = TypedSqliteTable(self._conn, model) 
        return self._tables[model]

    def commit(self):
        self._conn().commit()