fix: view decorator generic annotation.

src/sqlalchemy_declarative_extensions/dialects/sqlite/query.py

@@ -26,7 +26,9 @@ def check_schema_exists_sqlite(connection: Connection, name: str) -> bool:
 
 
 def get_views_sqlite(connection: Connection):
+    schemas = get_schemas_sqlite(connection)
     return [
         View(v.name, v.definition, schema=v.schema)
-        for v in connection.execute(views_query()).fetchall()
+        for schema in schemas
+        for v in connection.execute(views_query(schema.name)).fetchall()
     ]

src/sqlalchemy_declarative_extensions/dialects/sqlite/schema.py

@@ -1,28 +1,19 @@
 from typing import Optional
 
-from sqlalchemy import column, literal, table
+from sqlalchemy import bindparam, text
 
-from sqlalchemy_declarative_extensions.sqlalchemy import select
 
-
-def make_sqlite_schema(schema: Optional[str] = None):
-    tablename = "sqlite_schema"
+def views_query(schema: Optional[str] = None):
+    tablename = "sqlite_master"
     if schema:
         tablename = f"{schema}.{tablename}"
 
-    return table(
-        tablename,
-        column("type"),
-        column("name"),
-        column("sql"),
-    )
-
-
-def views_query(schema: Optional[str] = None):
-    sqlite_schema = make_sqlite_schema(schema)
-    return select(
-        literal(None).label("schema"),
-        literal(None),
-        sqlite_schema.c.name.label("name"),
-        sqlite_schema.c.sql.label("definition"),
-    ).where(sqlite_schema.c.type == "view")
+    return text(
+        "SELECT"  # noqa: S608
+        " :schema AS schema,"
+        " name AS name,"
+        " sql AS definition,"
+        " false as materialized"
+        f" FROM {tablename}"
+        " WHERE type == 'view'",
+    ).bindparams(bindparam("schema", schema))

src/sqlalchemy_declarative_extensions/view/base.py

@@ -22,9 +22,7 @@
 T = TypeVar("T")
 
 
-def view(
-    base: T, materialized: bool = False, register_as_model=False
-) -> Callable[[type], T]:
+def view(base, materialized: bool = False, register_as_model=False) -> Callable[[T], T]:
     """Decorate a class or declarative base model in order to register a View.
 
     Given some object with the attributes: `__tablename__`, (optionally for schema) `__table_args__`,

tests/view/test_update.py

@@ -60,10 +60,10 @@ def run_test(session):
     session.execute(text("CREATE VIEW bar AS SELECT id FROM foo WHERE id = 1"))
     session.execute(text("INSERT INTO foo (id) VALUES (1), (2), (12), (13)"))
 
-    result = [f.id for f in session.execute(text("SELECT id from bar")).all()]
+    result = [f.id for f in session.execute(text("SELECT id from bar")).fetchall()]
     assert result == [1]
 
     Base.metadata.create_all(bind=session.connection())
 
-    result = [f.id for f in session.execute(text("SELECT id from bar")).all()]
+    result = [f.id for f in session.execute(text("SELECT id from bar")).fetchall()]
     assert result == [1, 2]

tests/view/test_view_in_schema.py

@@ -0,0 +1,83 @@
+from pytest_mock_resources import (
+    create_postgres_fixture,
+    create_sqlite_fixture,
+)
+from sqlalchemy import Column, text, types
+
+from sqlalchemy_declarative_extensions import (
+    Row,
+    Rows,
+    Schemas,
+    View,
+    declarative_database,
+    register_sqlalchemy_events,
+    register_view,
+)
+from sqlalchemy_declarative_extensions.sqlalchemy import declarative_base
+
+_Base = declarative_base()
+
+
+@declarative_database
+class Base(_Base):  # type: ignore
+    __abstract__ = True
+
+    schemas = Schemas().are("fooschema")
+    rows = Rows().are(
+        Row("fooschema.foo", id=1),
+        Row("fooschema.foo", id=2),
+        Row("fooschema.foo", id=12),
+        Row("fooschema.foo", id=13),
+    )
+
+
+class Foo(Base):
+    __tablename__ = "foo"
+    __table_args__ = {"schema": "fooschema"}
+
+    id = Column(types.Integer(), primary_key=True)
+
+
+# Register imperitively
+view = View(
+    "bar",
+    "select id from fooschema.foo where id < 10",
+    schema="fooschema",
+)
+
+register_view(Base.metadata, view)
+
+
+register_sqlalchemy_events(Base.metadata, schemas=True, views=True, rows=True)
+
+pg = create_postgres_fixture(
+    scope="function", engine_kwargs={"echo": True}, session=True
+)
+sqlite = create_sqlite_fixture(scope="function", session=True)
+
+
+def test_create_view_postgresql(pg):
+    pg.execute(text("CREATE SCHEMA fooschema"))
+    run_test(pg)
+
+
+def test_create_view_sqlite(sqlite):
+    sqlite.execute(text("ATTACH DATABASE ':memory:' AS fooschema"))
+    run_test(sqlite)
+
+
+def run_test(session):
+    session.execute(text("CREATE TABLE fooschema.foo (id integer)"))
+    session.execute(
+        text("CREATE VIEW fooschema.bar AS SELECT id FROM fooschema.foo WHERE id = 1")
+    )
+    session.execute(text("INSERT INTO fooschema.foo (id) VALUES (1), (2), (12), (13)"))
+    session.commit()
+
+    result = [f.id for f in session.execute(text("SELECT id from fooschema.bar")).all()]
+    assert result == [1]
+
+    Base.metadata.create_all(bind=session.connection())
+
+    result = [f.id for f in session.execute(text("SELECT id from fooschema.bar")).all()]
+    assert result == [1, 2]