Merge pull request #20 from bdowning/updates

More updates

Author: bdowning

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.4.0-alpha-5
+current_version = 0.4.0-alpha-6
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(-(?P<release>.*)-(?P<build>\d+))?
 serialize = 
 	{major}.{minor}.{patch}-{release}-{build}

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "sql-athame"
-version = "0.4.0-alpha-5"
+version = "0.4.0-alpha-6"
 description = "Python tool for slicing and dicing SQL"
 authors = ["Brian Downing <bdowning@lavos.net>"]
 license = "MIT"

sql_athame/__init__.py

@@ -1,2 +1,2 @@
 from .base import Fragment, sql
-from .dataclasses import ModelBase, model_field, model_field_metadata
+from .dataclasses import ColumnInfo, ModelBase

sql_athame/dataclasses.py

@@ -1,52 +1,57 @@
 import datetime
 import uuid
-from collections.abc import AsyncGenerator, Iterable, Iterator, Mapping
-from dataclasses import dataclass, field, fields
+from collections.abc import AsyncGenerator, Iterable, Mapping
+from dataclasses import Field, InitVar, dataclass, fields
 from typing import (
+    Annotated,
     Any,
     Callable,
     Optional,
     TypeVar,
     Union,
+    get_origin,
+    get_type_hints,
 )
 
+from typing_extensions import TypeAlias
+
 from .base import Fragment, sql
 
-Where = Union[Fragment, Iterable[Fragment]]
+Where: TypeAlias = Union[Fragment, Iterable[Fragment]]
 # KLUDGE to avoid a string argument being valid
-SequenceOfStrings = Union[list[str], tuple[str, ...]]
-FieldNames = SequenceOfStrings
-FieldNamesSet = Union[SequenceOfStrings, set[str]]
+SequenceOfStrings: TypeAlias = Union[list[str], tuple[str, ...]]
+FieldNames: TypeAlias = SequenceOfStrings
+FieldNamesSet: TypeAlias = Union[SequenceOfStrings, set[str]]
 
-Connection = Any
-Pool = Any
+Connection: TypeAlias = Any
+Pool: TypeAlias = Any
 
 
 @dataclass
 class ColumnInfo:
     type: str
-    create_type: str
-    constraints: tuple[str, ...]
-
-    def create_table_string(self):
-        return " ".join((self.create_type, *self.constraints))
-
-
-def model_field_metadata(
-    type: str, constraints: Union[str, Iterable[str]] = ()
-) -> dict[str, Any]:
-    if isinstance(constraints, str):
-        constraints = (constraints,)
-    info = ColumnInfo(
-        sql_create_type_map.get(type.upper(), type), type, tuple(constraints)
-    )
-    return {"sql_athame": info}
-
-
-def model_field(
-    *, type: str, constraints: Union[str, Iterable[str]] = (), **kwargs: Any
-) -> Any:
-    return field(**kwargs, metadata=model_field_metadata(type, constraints))
+    create_type: str = ""
+    nullable: bool = False
+    _constraints: tuple[str, ...] = ()
+
+    constraints: InitVar[Union[str, Iterable[str], None]] = None
+
+    def __post_init__(self, constraints: Union[str, Iterable[str], None]) -> None:
+        if self.create_type == "":
+            self.create_type = self.type
+            self.type = sql_create_type_map.get(self.type.upper(), self.type)
+        if constraints is not None:
+            if type(constraints) is str:
+                constraints = (constraints,)
+            self._constraints = tuple(constraints)
+
+    def create_table_string(self) -> str:
+        parts = (
+            self.create_type,
+            *(() if self.nullable else ("NOT NULL",)),
+            *self._constraints,
+        )
+        return " ".join(parts)
 
 
 sql_create_type_map = {
@@ -56,43 +61,37 @@ def model_field(
 }
 
 
-sql_type_map = {
-    Optional[bool]: ("BOOLEAN",),
-    Optional[bytes]: ("BYTEA",),
-    Optional[datetime.date]: ("DATE",),
-    Optional[datetime.datetime]: ("TIMESTAMP",),
-    Optional[float]: ("DOUBLE PRECISION",),
-    Optional[int]: ("INTEGER",),
-    Optional[str]: ("TEXT",),
-    Optional[uuid.UUID]: ("UUID",),
-    bool: ("BOOLEAN", "NOT NULL"),
-    bytes: ("BYTEA", "NOT NULL"),
-    datetime.date: ("DATE", "NOT NULL"),
-    datetime.datetime: ("TIMESTAMP", "NOT NULL"),
-    float: ("DOUBLE PRECISION", "NOT NULL"),
-    int: ("INTEGER", "NOT NULL"),
-    str: ("TEXT", "NOT NULL"),
-    uuid.UUID: ("UUID", "NOT NULL"),
+sql_type_map: dict[Any, tuple[str, bool]] = {
+    Optional[bool]: ("BOOLEAN", True),
+    Optional[bytes]: ("BYTEA", True),
+    Optional[datetime.date]: ("DATE", True),
+    Optional[datetime.datetime]: ("TIMESTAMP", True),
+    Optional[float]: ("DOUBLE PRECISION", True),
+    Optional[int]: ("INTEGER", True),
+    Optional[str]: ("TEXT", True),
+    Optional[uuid.UUID]: ("UUID", True),
+    bool: ("BOOLEAN", False),
+    bytes: ("BYTEA", False),
+    datetime.date: ("DATE", False),
+    datetime.datetime: ("TIMESTAMP", False),
+    float: ("DOUBLE PRECISION", False),
+    int: ("INTEGER", False),
+    str: ("TEXT", False),
+    uuid.UUID: ("UUID", False),
 }
 
 
-def column_info_for_field(field):
-    if "sql_athame" in field.metadata:
-        return field.metadata["sql_athame"]
-    type, *constraints = sql_type_map[field.type]
-    return ColumnInfo(type, type, tuple(constraints))
-
-
 T = TypeVar("T", bound="ModelBase")
 U = TypeVar("U")
 
 
-class ModelBase(Mapping[str, Any]):
+class ModelBase:
     _column_info: Optional[dict[str, ColumnInfo]]
     _cache: dict[tuple, Any]
     table_name: str
     primary_key_names: tuple[str, ...]
     array_safe_insert: bool
+    _type_hints: dict[str, type]
 
     def __init_subclass__(
         cls,
@@ -130,27 +129,34 @@ def _cached(cls, key: tuple, thunk: Callable[[], U]) -> U:
             cls._cache[key] = thunk()
             return cls._cache[key]
 
-    def keys(self):
-        return self.field_names()
-
-    def __getitem__(self, key: str) -> Any:
-        return getattr(self, key)
-
-    def __iter__(self) -> Iterator[Any]:
-        return iter(self.keys())
-
-    def __len__(self) -> int:
-        return len(self.keys())
+    @classmethod
+    def type_hints(cls) -> dict[str, type]:
+        try:
+            return cls._type_hints
+        except AttributeError:
+            cls._type_hints = get_type_hints(cls, include_extras=True)
+            return cls._type_hints
 
-    def get(self, key: str, default: Any = None) -> Any:
-        return getattr(self, key, default)
+    @classmethod
+    def column_info_for_field(cls, field: Field) -> ColumnInfo:
+        type_info = cls.type_hints()[field.name]
+        base_type = type_info
+        if get_origin(type_info) is Annotated:
+            base_type = type_info.__origin__  # type: ignore
+            for md in type_info.__metadata__:  # type: ignore
+                if isinstance(md, ColumnInfo):
+                    return md
+        type, nullable = sql_type_map[base_type]
+        return ColumnInfo(type=type, nullable=nullable)
 
     @classmethod
     def column_info(cls, column: str) -> ColumnInfo:
         try:
             return cls._column_info[column]  # type: ignore
         except AttributeError:
-            cls._column_info = {f.name: column_info_for_field(f) for f in cls._fields()}
+            cls._column_info = {
+                f.name: cls.column_info_for_field(f) for f in cls._fields()
+            }
             return cls._column_info[column]
 
     @classmethod

sql_athame/types.py

@@ -1,6 +1,8 @@
 import dataclasses
 from typing import TYPE_CHECKING, Any, Union
 
+from typing_extensions import TypeAlias
+
 
 @dataclasses.dataclass(eq=False)
 class Placeholder:
@@ -15,8 +17,8 @@ class Slot:
     name: str
 
 
-Part = Union[str, Placeholder, Slot, "Fragment"]
-FlatPart = Union[str, Placeholder, Slot]
+Part: TypeAlias = Union[str, Placeholder, Slot, "Fragment"]
+FlatPart: TypeAlias = Union[str, Placeholder, Slot]
 
 if TYPE_CHECKING:
     from .base import Fragment

tests/test_asyncpg.py

@@ -1,14 +1,19 @@
+# ruff: noqa: UP007
+
+from __future__ import annotations
+
 import asyncio
 import json
 import os
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Optional
+from typing import Annotated, Optional
 
 import asyncpg
 import pytest
+from typing_extensions import TypeAlias
 
-from sql_athame import ModelBase, model_field_metadata, sql
+from sql_athame import ColumnInfo, ModelBase, sql
 
 
 @pytest.fixture(autouse=True)
@@ -137,9 +142,7 @@ class Test(
         insert_multiple_mode="array_safe",
     ):
         id: int
-        a: list[int] = field(
-            metadata=model_field_metadata(type="INT[]", constraints="NOT NULL")
-        )
+        a: Annotated[list[int], ColumnInfo(type="INT[]")]
         b: str
 
     await conn.execute(*Test.create_table_sql())
@@ -262,10 +265,13 @@ class Test(ModelBase, table_name="test", primary_key=("id1", "id2")):
     ]
 
 
+Serial: TypeAlias = Annotated[int, ColumnInfo(type="SERIAL")]
+
+
 async def test_serial(conn):
     @dataclass
     class Test(ModelBase, table_name="table", primary_key="id"):
-        id: int = field(metadata=model_field_metadata(type="SERIAL"))
+        id: Serial
         foo: int
         bar: str
 
@@ -281,8 +287,8 @@ class Test(ModelBase, table_name="table", primary_key="id"):
 async def test_unnest_json(conn):
     @dataclass
     class Test(ModelBase, table_name="table", primary_key="id"):
-        id: int = field(metadata=model_field_metadata(type="SERIAL"))
-        json: Optional[list] = field(metadata=model_field_metadata(type="JSONB"))
+        id: Serial
+        json: Annotated[Optional[list], ColumnInfo(type="JSONB", nullable=True)]
 
     await conn.set_type_codec(
         "jsonb", encoder=json.dumps, decoder=json.loads, schema="pg_catalog"
@@ -307,7 +313,7 @@ class Test(ModelBase, table_name="table", primary_key="id"):
 async def test_unnest_empty(conn):
     @dataclass
     class Test(ModelBase, table_name="table", primary_key="id"):
-        id: int = field(metadata=model_field_metadata(type="SERIAL"))
+        id: Serial
 
     await conn.execute(*Test.create_table_sql())
 

tests/test_dataclasses.py

@@ -1,21 +1,20 @@
+# ruff: noqa: UP007
+
+from __future__ import annotations
+
 import uuid
-from dataclasses import dataclass, field
-from typing import Optional
+from dataclasses import dataclass
+from typing import Annotated, Optional
 
 from sql_athame import sql
-from sql_athame.dataclasses import ModelBase, model_field_metadata
+from sql_athame.dataclasses import ColumnInfo, ModelBase
 
 
 def test_modelclass():
     @dataclass
     class Test(ModelBase, table_name="table"):
-        foo: int = field(
-            metadata=model_field_metadata(type="INTEGER", constraints="NOT NULL")
-        )
-        bar: str = field(
-            default="hi",
-            metadata=model_field_metadata(type="TEXT", constraints="NOT NULL"),
-        )
+        foo: int
+        bar: str = "hi"
 
     t = Test(42)
 
@@ -96,36 +95,18 @@ class Test(ModelBase, table_name="table", primary_key="id"):
     ]
 
 
-def test_mapping():
-    @dataclass
-    class Test(ModelBase, table_name="table", primary_key="id"):
-        id: int
-        foo: int
-        bar: str
-
-    t = Test(1, 2, "foo")
-    assert t["id"] == 1
-    assert t["foo"] == 2
-    assert t["bar"] == "foo"
-
-    assert list(t.keys()) == ["id", "foo", "bar"]
-
-    assert dict(t) == {"id": 1, "foo": 2, "bar": "foo"}
-    assert dict(**t) == {"id": 1, "foo": 2, "bar": "foo"}
-
-
 def test_serial():
     @dataclass
     class Test(ModelBase, table_name="table", primary_key="id"):
-        id: int = field(metadata=model_field_metadata(type="SERIAL"))
+        id: Annotated[int, ColumnInfo(type="SERIAL")]
         foo: int
         bar: str
 
     assert Test.column_info("id").type == "INTEGER"
     assert Test.column_info("id").create_type == "SERIAL"
     assert list(Test.create_table_sql()) == [
         'CREATE TABLE IF NOT EXISTS "table" ('
-        '"id" SERIAL, '
+        '"id" SERIAL NOT NULL, '
         '"foo" INTEGER NOT NULL, '
         '"bar" TEXT NOT NULL, '
         'PRIMARY KEY ("id"))'