Add sqlalchemy example

Author: penberg

examples/sqlalchemy/dialect.py

@@ -0,0 +1,119 @@
+import os
+import urllib.parse
+
+from sqlalchemy import util
+from sqlalchemy.dialects import registry as _registry
+from sqlalchemy.dialects.sqlite.pysqlite import SQLiteDialect_pysqlite
+
+__version__ = "0.1.0-pre"
+
+_registry.register(
+    "sqlite.libsql", "sqlalchemy_libsql", "SQLiteDialect_libsql"
+)
+
+
+def _build_connection_url(url, query, secure):
+    # sorting of keys is for unit test support
+    query_str = urllib.parse.urlencode(sorted(query.items()))
+
+    if not url.host:
+        if query_str:
+            return f"{url.database}?{query_str}"
+        return url.database
+    elif secure:  # yes, pop to remove
+        scheme = "wss"
+    else:
+        scheme = "ws"
+
+    if url.username and url.password:
+        netloc = f"{url.username}:{url.password}@{url.host}"
+    elif url.username:
+        netloc = f"{url.username}@{url.host}"
+    else:
+        netloc = url.host
+
+    if url.port:
+        netloc += f":{url.port}"
+
+    return urllib.parse.urlunsplit(
+        (
+            scheme,
+            netloc,
+            url.database or "",
+            query_str,
+            "",  # fragment
+        )
+    )
+
+
+class SQLiteDialect_libsql(SQLiteDialect_pysqlite):
+    driver = "libsql"
+    # need to be set explicitly
+    supports_statement_cache = SQLiteDialect_pysqlite.supports_statement_cache
+
+    @classmethod
+    def import_dbapi(cls):
+        import libsql_experimental as libsql
+
+        return libsql
+
+    def on_connect(self):
+        import libsql_experimental as libsql
+
+        sqlite3_connect = super().on_connect()
+
+        def connect(conn):
+            # LibSQL: there is no support for create_function()
+            if isinstance(conn, Connection):
+                return
+            return sqlite3_connect(conn)
+
+        return connect
+
+    def create_connect_args(self, url):
+        pysqlite_args = (
+            ("uri", bool),
+            ("timeout", float),
+            ("isolation_level", str),
+            ("detect_types", int),
+            ("check_same_thread", bool),
+            ("cached_statements", int),
+            ("secure", bool),  # LibSQL extra, selects between ws and wss
+        )
+        opts = url.query
+        libsql_opts = {}
+        for key, type_ in pysqlite_args:
+            util.coerce_kw_type(opts, key, type_, dest=libsql_opts)
+
+        if url.host:
+            libsql_opts["uri"] = True
+
+        if libsql_opts.get("uri", False):
+            uri_opts = dict(opts)
+            # here, we are actually separating the parameters that go to
+            # sqlite3/pysqlite vs. those that go the SQLite URI.  What if
+            # two names conflict?  again, this seems to be not the case right
+            # now, and in the case that new names are added to
+            # either side which overlap, again the sqlite3/pysqlite parameters
+            # can be passed through connect_args instead of in the URL.
+            # If SQLite native URIs add a parameter like "timeout" that
+            # we already have listed here for the python driver, then we need
+            # to adjust for that here.
+            for key, type_ in pysqlite_args:
+                uri_opts.pop(key, None)
+
+            secure = libsql_opts.pop("secure", False)
+            connect_url = _build_connection_url(url, uri_opts, secure)
+        else:
+            connect_url = url.database or ":memory:"
+            if connect_url != ":memory:":
+                connect_url = os.path.abspath(connect_url)
+
+        libsql_opts.setdefault(
+            "check_same_thread", not self._is_url_file_db(url)
+        )
+
+        return ([connect_url], libsql_opts)
+
+
+dialect = SQLiteDialect_libsql

examples/sqlalchemy/example.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+
+import dialect
+
+from typing import List
+from typing import Optional
+from sqlalchemy import ForeignKey
+from sqlalchemy import String
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
+from sqlalchemy.orm import relationship
+
+class Base(DeclarativeBase):
+    pass
+
+class User(Base):
+    __tablename__ = "user_account"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    name: Mapped[str] = mapped_column(String(30))
+    fullname: Mapped[Optional[str]]
+    addresses: Mapped[List["Address"]] = relationship(
+        back_populates="user", cascade="all, delete-orphan"
+    )
+    def __repr__(self) -> str:
+        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"
+
+class Address(Base):
+    __tablename__ = "address"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    email_address: Mapped[str]
+    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
+    user: Mapped["User"] = relationship(back_populates="addresses")
+    def __repr__(self) -> str:
+        return f"Address(id={self.id!r}, email_address={self.email_address!r})"
+
+from sqlalchemy import create_engine
+engine = create_engine("sqlite+libsql://", echo=True)
+
+Base.metadata.create_all(engine)
+
+from sqlalchemy.orm import Session
+
+with Session(engine) as session:
+    spongebob = User(
+        name="spongebob",
+        fullname="Spongebob Squarepants",
+        addresses=[Address(email_address="[email protected]")],
+    )
+    sandy = User(
+        name="sandy",
+        fullname="Sandy Cheeks",
+        addresses=[
+            Address(email_address="[email protected]"),
+            Address(email_address="[email protected]"),
+        ],
+    )
+    patrick = User(name="patrick", fullname="Patrick Star")
+    session.add_all([spongebob, sandy, patrick])
+    session.commit()
+
+from sqlalchemy import select
+
+session = Session(engine)
+
+stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))
+
+for user in session.scalars(stmt):
+    print(user)