simulation

Author: joeblackwaslike

src/quart_sqlalchemy/model/mixins.py

@@ -79,7 +79,7 @@ def to_dict(self):
 class RecursiveDictMixin:
     __abstract__ = True
 
-    def model_to_dict(
+    def to_dict(
         self,
         obj: t.Optional[t.Any] = None,
         max_depth: int = 3,

src/quart_sqlalchemy/model/model.py

@@ -14,13 +14,14 @@
 from .mixins import ComparableMixin
 from .mixins import DynamicArgsMixin
 from .mixins import ReprMixin
+from .mixins import SimpleDictMixin
 from .mixins import TableNameMixin
 
 
 sa = sqlalchemy
 
 
-class Base(DynamicArgsMixin, ReprMixin, ComparableMixin, TableNameMixin):
+class Base(DynamicArgsMixin, ReprMixin, SimpleDictMixin, ComparableMixin, TableNameMixin):
     __abstract__ = True
 
     type_annotation_map = {

src/quart_sqlalchemy/sim/__init__.py

@@ -0,0 +1,2 @@
+from . import app
+from . import model

src/quart_sqlalchemy/sim/app.py

@@ -0,0 +1,145 @@
+import json
+import logging
+import re
+import typing as t
+
+import sqlalchemy as sa
+from pydantic import BaseModel
+from quart import g
+from quart import Quart
+from quart import request
+from quart import Request
+from quart import Response
+from quart_schema import QuartSchema
+
+from .. import Base
+from .. import SQLAlchemyConfig
+from ..framework import QuartSQLAlchemy
+from .util import ObjectID
+
+
+AUTHORIZATION_PATTERN = re.compile(r"Bearer (?P<token>.+)")
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class MyBase(Base):
+    type_annotation_map = {ObjectID: sa.Integer}
+
+
+app = Quart(__name__)
+db = QuartSQLAlchemy(
+    SQLAlchemyConfig.parse_obj(
+        {
+            "model_class": MyBase,
+            "binds": {
+                "default": {
+                    "engine": {"url": "sqlite:///file:mem.db?mode=memory&cache=shared&uri=true"},
+                    "session": {"expire_on_commit": False},
+                },
+                "read-replica": {
+                    "engine": {"url": "sqlite:///file:mem.db?mode=memory&cache=shared&uri=true"},
+                    "session": {"expire_on_commit": False},
+                    "read_only": True,
+                },
+                "async": {
+                    "engine": {
+                        "url": "sqlite+aiosqlite:///file:mem.db?mode=memory&cache=shared&uri=true"
+                    },
+                    "session": {"expire_on_commit": False},
+                },
+            },
+        }
+    )
+)
+openapi = QuartSchema(app)
+
+
+class RequestAuth(BaseModel):
+    client: t.Optional[t.Any] = None
+    user: t.Optional[t.Any] = None
+
+    @property
+    def has_client(self):
+        return self.client is not None
+
+    @property
+    def has_user(self):
+        return self.user is not None
+
+    @property
+    def is_anonymous(self):
+        return all([self.has_client is False, self.has_user is False])
+
+
+def get_request_client(request: Request):
+    api_key = request.headers.get("X-Public-API-Key")
+    if not api_key:
+        return
+
+    with g.bind.Session() as session:
+        try:
+            magic_client = g.h.MagicClient(session).get_by_public_api_key(api_key)
+        except ValueError:
+            return
+        else:
+            return magic_client
+
+
+def get_request_user(request: Request):
+    auth_header = request.headers.get("Authorization")
+
+    if not auth_header:
+        return
+    m = AUTHORIZATION_PATTERN.match(auth_header)
+    if m is None:
+        raise RuntimeError("invalid authorization header")
+
+    auth_token = m.group("auth_token")
+
+    with g.bind.Session() as session:
+        try:
+            auth_user = g.h.AuthUser(session).get_by_session_token(auth_token)
+        except ValueError:
+            return
+        else:
+            return auth_user
+
+
+@app.before_request
+def set_ethereum_network():
+    g.request_network = request.headers.get("X-Fortmatic-Network", "GOERLI").upper()
+
+
+@app.before_request
+def set_bind_handlers_for_request():
+    from quart_sqlalchemy.sim.handle import Handlers
+
+    g.db = db
+
+    method = request.method
+    if method in ["GET", "OPTIONS", "TRACE", "HEAD"]:
+        bind = "read-replica"
+    else:
+        bind = "default"
+
+    g.bind = db.get_bind(bind)
+    g.h = Handlers(g.bind)
+
+
+@app.before_request
+def set_request_auth():
+    g.auth = RequestAuth(
+        client=get_request_client(request),
+        user=get_request_user(request),
+    )
+
+
+@app.after_request
+async def add_json_response_envelope(response: Response) -> Response:
+    if response.mimetype != "application/json":
+        return response
+    data = await response.get_json()
+    payload = dict(status="ok", message="", data=data)
+    response.set_data(json.dumps(payload))
+    return response

src/quart_sqlalchemy/sim/builder.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+import typing as t
+
+import sqlalchemy
+import sqlalchemy.event
+import sqlalchemy.exc
+import sqlalchemy.orm
+import sqlalchemy.sql
+from sqlalchemy.orm.interfaces import ORMOption
+
+from quart_sqlalchemy.types import ColumnExpr
+from quart_sqlalchemy.types import DMLTable
+from quart_sqlalchemy.types import EntityT
+from quart_sqlalchemy.types import Selectable
+
+
+sa = sqlalchemy
+
+
+class StatementBuilder(t.Generic[EntityT]):
+    model: t.Type[EntityT]
+
+    def __init__(self, model: t.Type[EntityT]):
+        self.model = model
+
+    def complex_select(
+        self,
+        selectables: t.Sequence[Selectable] = (),
+        conditions: t.Sequence[ColumnExpr] = (),
+        group_by: t.Sequence[t.Union[ColumnExpr, str]] = (),
+        order_by: t.Sequence[t.Union[ColumnExpr, str]] = (),
+        options: t.Sequence[ORMOption] = (),
+        execution_options: t.Optional[t.Dict[str, t.Any]] = None,
+        offset: t.Optional[int] = None,
+        limit: t.Optional[int] = None,
+        distinct: bool = False,
+        for_update: bool = False,
+    ) -> sa.Select:
+        statement = sa.select(*selectables or self.model).where(*conditions)
+
+        if for_update:
+            statement = statement.with_for_update()
+        if offset:
+            statement = statement.offset(offset)
+        if limit:
+            statement = statement.limit(limit)
+        if group_by:
+            statement = statement.group_by(*group_by)
+        if order_by:
+            statement = statement.order_by(*order_by)
+
+        for option in options:
+            for context in option.context:
+                for strategy in context.strategy:
+                    if "joined" in strategy:
+                        distinct = True
+
+            statement = statement.options(option)
+
+        if distinct:
+            statement = statement.distinct()
+
+        if execution_options:
+            statement = statement.execution_options(**execution_options)
+
+        return statement
+
+    def insert(
+        self,
+        target: t.Optional[DMLTable] = None,
+        values: t.Optional[t.Dict[str, t.Any]] = None,
+    ) -> sa.Insert:
+        return sa.insert(target or self.model).values(**values or {})
+
+    def bulk_insert(
+        self,
+        target: t.Optional[DMLTable] = None,
+        values: t.Sequence[t.Dict[str, t.Any]] = (),
+    ) -> sa.Insert:
+        return sa.insert(target or self.model).values(*values)
+
+    def bulk_update(
+        self,
+        target: t.Optional[DMLTable] = None,
+        conditions: t.Sequence[ColumnExpr] = (),
+        values: t.Optional[t.Dict[str, t.Any]] = None,
+    ) -> sa.Update:
+        return sa.update(target or self.model).where(*conditions).values(**values or {})
+
+    def bulk_delete(
+        self,
+        target: t.Optional[DMLTable] = None,
+        conditions: t.Sequence[ColumnExpr] = (),
+    ) -> sa.Delete:
+        return sa.delete(target or self.model).where(*conditions)