We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
StaticFilesConfig and virtual directories
I'm trying to write a ``FileSystemProtocol`` to load files from the package data using [importlib_resources](https://importlib-resources.readthedocs.io/en/latest/using.html#). But because ``directories`` is defined as ``DirectoryPath``, pydantic checks if the given directories exist in the local filesystem. 

This is not generally true, especially in any kind of virtual filesystem (e.g. a zipped package). I think this condition should be relaxed to support virtual filesystems.

https://github.com/starlite-api/starlite/blob/9bb6dcd57c10a591377cf8e3a537e9292566d5b9/starlite/config/static_files.py#L32

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `litestar/contrib/sqlalchemy/base.py`
Content:
```
1 """Application ORM configuration."""
2 from __future__ import annotations
3 
4 import re
5 from datetime import date, datetime
6 from typing import TYPE_CHECKING, Any, ClassVar, Protocol, TypeVar, runtime_checkable
7 from uuid import UUID, uuid4
8 
9 from pydantic import AnyHttpUrl, AnyUrl, EmailStr
10 from sqlalchemy import Date, DateTime, MetaData, Sequence, String
11 from sqlalchemy.event import listens_for
12 from sqlalchemy.orm import (
13     DeclarativeBase,
14     Mapped,
15     Session,
16     declared_attr,
17     mapped_column,
18     orm_insert_sentinel,
19     registry,
20 )
21 
22 from .types import GUID, BigIntIdentity, JsonB
23 
24 if TYPE_CHECKING:
25     from sqlalchemy.sql import FromClause
26 
27 __all__ = (
28     "AuditColumns",
29     "BigIntAuditBase",
30     "BigIntBase",
31     "BigIntPrimaryKey",
32     "CommonTableAttributes",
33     "create_registry",
34     "ModelProtocol",
35     "touch_updated_timestamp",
36     "UUIDAuditBase",
37     "UUIDBase",
38     "UUIDPrimaryKey",
39 )
40 
41 
42 UUIDBaseT = TypeVar("UUIDBaseT", bound="UUIDBase")
43 BigIntBaseT = TypeVar("BigIntBaseT", bound="BigIntBase")
44 
45 convention = {
46     "ix": "ix_%(column_0_label)s",
47     "uq": "uq_%(table_name)s_%(column_0_name)s",
48     "ck": "ck_%(table_name)s_%(constraint_name)s",
49     "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
50     "pk": "pk_%(table_name)s",
51 }
52 """Templates for automated constraint name generation."""
53 
54 
55 @listens_for(Session, "before_flush")
56 def touch_updated_timestamp(session: Session, *_: Any) -> None:
57     """Set timestamp on update.
58 
59     Called from SQLAlchemy's
60     :meth:`before_flush <sqlalchemy.orm.SessionEvents.before_flush>` event to bump the ``updated``
61     timestamp on modified instances.
62 
63     Args:
64         session: The sync :class:`Session <sqlalchemy.orm.Session>` instance that underlies the async
65             session.
66     """
67     for instance in session.dirty:
68         if hasattr(instance, "updated"):
69             instance.updated = datetime.now()  # noqa: DTZ005
70 
71 
72 @runtime_checkable
73 class ModelProtocol(Protocol):
74     """The base SQLAlchemy model protocol."""
75 
76     __table__: FromClause
77     __name__: ClassVar[str]
78 
79     def to_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
80         """Convert model to dictionary.
81 
82         Returns:
83             dict[str, Any]: A dict representation of the model
84         """
85         ...
86 
87 
88 class UUIDPrimaryKey:
89     """UUID Primary Key Field Mixin."""
90 
91     id: Mapped[UUID] = mapped_column(default=uuid4, primary_key=True)  # pyright: ignore
92     """UUID Primary key column."""
93 
94     @declared_attr
95     def _sentinel(cls) -> Mapped[int]:
96         return orm_insert_sentinel()
97 
98 
99 class BigIntPrimaryKey:
100     """BigInt Primary Key Field Mixin."""
101 
102     @declared_attr
103     def id(cls) -> Mapped[int]:
104         """BigInt Primary key column."""
105         return mapped_column(
106             BigIntIdentity,
107             Sequence(f"{cls.__tablename__}_id_seq", optional=False),  # type: ignore[attr-defined] # pyright: ignore
108             primary_key=True,
109         )
110 
111 
112 class AuditColumns:
113     """Created/Updated At Fields Mixin."""
114 
115     created: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore
116     """Date/time of instance creation."""
117     updated: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore
118     """Date/time of instance last update."""
119 
120 
121 class CommonTableAttributes:
122     """Common attributes for SQLALchemy tables."""
123 
124     __name__: ClassVar[str]
125     __table__: FromClause
126 
127     # noinspection PyMethodParameters
128     @declared_attr.directive
129     def __tablename__(cls) -> str:  # pylint: disable=no-self-argument
130         """Infer table name from class name."""
131         regexp = re.compile("((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))")
132         return regexp.sub(r"_\1", cls.__name__).lower()
133 
134     def to_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:
135         """Convert model to dictionary.
136 
137         Returns:
138             dict[str, Any]: A dict representation of the model
139         """
140         exclude = {"_sentinel"}.union(self._sa_instance_state.unloaded).union(exclude or [])  # type: ignore[attr-defined]
141         return {field.name: getattr(self, field.name) for field in self.__table__.columns if field.name not in exclude}
142 
143 
144 def create_registry() -> registry:
145     """Create a new SQLAlchemy registry."""
146     meta = MetaData(naming_convention=convention)
147     return registry(
148         metadata=meta,
149         type_annotation_map={
150             UUID: GUID,
151             EmailStr: String,
152             AnyUrl: String,
153             AnyHttpUrl: String,
154             dict: JsonB,
155             datetime: DateTime,
156             date: Date,
157         },
158     )
159 
160 
161 orm_registry = create_registry()
162 
163 
164 class UUIDBase(UUIDPrimaryKey, CommonTableAttributes, DeclarativeBase):
165     """Base for all SQLAlchemy declarative models with UUID primary keys."""
166 
167     registry = orm_registry
168 
169 
170 class UUIDAuditBase(CommonTableAttributes, UUIDPrimaryKey, AuditColumns, DeclarativeBase):
171     """Base for declarative models with UUID primary keys and audit columns."""
172 
173     registry = orm_registry
174 
175 
176 class BigIntBase(BigIntPrimaryKey, CommonTableAttributes, DeclarativeBase):
177     """Base for all SQLAlchemy declarative models with BigInt primary keys."""
178 
179     registry = orm_registry
180 
181 
182 class BigIntAuditBase(CommonTableAttributes, BigIntPrimaryKey, AuditColumns, DeclarativeBase):
183     """Base for declarative models with BigInt primary keys and audit columns."""
184 
185     registry = orm_registry
186
```

Path: `litestar/contrib/sqlalchemy/types.py`
Content:
```
1 from __future__ import annotations
2 
3 import uuid
4 from base64 import b64decode
5 from typing import TYPE_CHECKING, Any, cast
6 
7 from sqlalchemy import text, util
8 from sqlalchemy.dialects.oracle import BLOB as ORA_BLOB
9 from sqlalchemy.dialects.oracle import RAW as ORA_RAW
10 from sqlalchemy.dialects.postgresql import JSONB as PG_JSONB
11 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
12 from sqlalchemy.types import BINARY, CHAR, BigInteger, Integer, SchemaType, TypeDecorator, TypeEngine
13 from sqlalchemy.types import JSON as _JSON
14 
15 from litestar.serialization import decode_json, encode_json
16 
17 if TYPE_CHECKING:
18     from sqlalchemy.engine import Dialect
19 
20 
21 class GUID(TypeDecorator):
22     """Platform-independent GUID type.
23 
24     Uses PostgreSQL's UUID type, Oracle's RAW(16) type, otherwise uses
25     BINARY(16) or CHAR(32), storing as stringified hex values.
26 
27     Will accept stringified UUIDs as a hexstring or an actual UUID
28 
29     """
30 
31     impl = BINARY(16)
32     cache_ok = True
33 
34     @property
35     def python_type(self) -> type[uuid.UUID]:
36         return uuid.UUID
37 
38     def __init__(self, *args: Any, binary: bool = True, **kwargs: Any) -> None:
39         self.binary = binary
40 
41     def load_dialect_impl(self, dialect: Dialect) -> Any:
42         if dialect.name in {"postgresql", "duckdb"}:
43             return dialect.type_descriptor(PG_UUID())
44         if dialect.name == "oracle":
45             return dialect.type_descriptor(ORA_RAW(16))
46         if self.binary:
47             return dialect.type_descriptor(BINARY(16))
48         return dialect.type_descriptor(CHAR(32))
49 
50     def process_bind_param(self, value: bytes | str | uuid.UUID | None, dialect: Dialect) -> bytes | str | None:
51         if value is None:
52             return value
53         if dialect.name in {"postgresql", "duckdb"}:
54             return str(value)
55         value = self.to_uuid(value)
56         if value is None:
57             return value
58         if dialect.name in {"oracle", "spanner+spanner"}:
59             return value.bytes
60         return value.bytes if self.binary else value.hex
61 
62     def process_result_value(self, value: bytes | str | uuid.UUID | None, dialect: Dialect) -> uuid.UUID | None:
63         if value is None:
64             return value
65         if isinstance(value, uuid.UUID):
66             return value
67         if dialect.name == "spanner+spanner":
68             return uuid.UUID(bytes=b64decode(value))
69         if self.binary:
70             return uuid.UUID(bytes=cast("bytes", value))
71         return uuid.UUID(hex=cast("str", value))
72 
73     @staticmethod
74     def to_uuid(value: Any) -> uuid.UUID | None:
75         if isinstance(value, uuid.UUID) or value is None:
76             return value
77         try:
78             value = uuid.UUID(hex=value)
79         except (TypeError, ValueError):
80             value = uuid.UUID(bytes=value)
81         return cast("uuid.UUID | None", value)
82 
83 
84 class ORA_JSONB(TypeDecorator, SchemaType):  # type: ignore  # noqa: N801
85     """Oracle Binary JSON type.
86 
87     JsonB = _JSON().with_variant(PG_JSONB, "postgresql").with_variant(ORA_JSONB, "oracle")
88 
89     """
90 
91     impl = ORA_BLOB
92     cache_ok = True
93 
94     @property
95     def python_type(self) -> type[dict]:
96         return dict
97 
98     def __init__(self, *args: Any, **kwargs: Any) -> None:
99         """Initialize JSON type"""
100         self.name = kwargs.pop("name", None)
101         self.oracle_strict = kwargs.pop("oracle_strict", True)
102 
103     def coerce_compared_value(self, op: Any, value: Any) -> Any:
104         return self.impl.coerce_compared_value(op=op, value=value)  # type: ignore
105 
106     def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:
107         return dialect.type_descriptor(ORA_BLOB())
108 
109     def process_bind_param(self, value: Any, dialect: Dialect) -> Any | None:
110         if value is None:
111             return value
112         return encode_json(value)
113 
114     def process_result_value(self, value: bytes | None, dialect: Dialect) -> Any | None:
115         if value is None:
116             return value
117         return decode_json(value)
118 
119     def _should_create_constraint(self, compiler: Any, **kw: Any) -> bool:
120         return bool(compiler.dialect.name == "oracle")
121 
122     def _variant_mapping_for_set_table(self, column: Any) -> dict | None:
123         if column.type._variant_mapping:
124             variant_mapping = dict(column.type._variant_mapping)
125             variant_mapping["_default"] = column.type
126         else:
127             variant_mapping = None
128         return variant_mapping
129 
130     @util.preload_module("sqlalchemy.sql.schema")
131     def _set_table(self, column: Any, table: Any) -> None:
132         schema = util.preloaded.sql_schema
133         variant_mapping = self._variant_mapping_for_set_table(column)
134         constraint_options = "(strict)" if self.oracle_strict else ""
135         sqltext = text(f"{column.name} is json {constraint_options}")
136         e = schema.CheckConstraint(
137             sqltext,
138             name=f"{column.name}_is_json",
139             _create_rule=util.portable_instancemethod(  # type: ignore[no-untyped-call]
140                 self._should_create_constraint,
141                 {"variant_mapping": variant_mapping},
142             ),
143             _type_bound=True,
144         )
145         table.append_constraint(e)
146 
147 
148 BigIntIdentity = BigInteger().with_variant(Integer, "sqlite")
149 """A ``BigInteger`` variant that reverts to an ``Integer`` for unsupported variants."""
150 
151 JsonB = _JSON().with_variant(PG_JSONB, "postgresql").with_variant(ORA_JSONB, "oracle")
152 """A JSON type that uses  native ``JSONB`` where possible and ``Binary`` or ``Blob`` as
153 an alternative.
154 """
155
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/litestar/contrib/sqlalchemy/base.py b/litestar/contrib/sqlalchemy/base.py
--- a/litestar/contrib/sqlalchemy/base.py
+++ b/litestar/contrib/sqlalchemy/base.py
@@ -2,12 +2,12 @@
 from __future__ import annotations
 
 import re
-from datetime import date, datetime
+from datetime import date, datetime, timezone
 from typing import TYPE_CHECKING, Any, ClassVar, Protocol, TypeVar, runtime_checkable
 from uuid import UUID, uuid4
 
 from pydantic import AnyHttpUrl, AnyUrl, EmailStr
-from sqlalchemy import Date, DateTime, MetaData, Sequence, String
+from sqlalchemy import Date, MetaData, Sequence, String
 from sqlalchemy.event import listens_for
 from sqlalchemy.orm import (
     DeclarativeBase,
@@ -19,7 +19,7 @@
     registry,
 )
 
-from .types import GUID, BigIntIdentity, JsonB
+from .types import GUID, BigIntIdentity, DateTimeUTC, JsonB
 
 if TYPE_CHECKING:
     from sqlalchemy.sql import FromClause
@@ -65,8 +65,8 @@
             session.
     """
     for instance in session.dirty:
-        if hasattr(instance, "updated"):
-            instance.updated = datetime.now()  # noqa: DTZ005
+        if hasattr(instance, "updated_at"):
+            instance.updated = (datetime.now(timezone.utc),)
 
 
 @runtime_checkable
@@ -112,9 +112,15 @@
 class AuditColumns:
     """Created/Updated At Fields Mixin."""
 
-    created: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore
+    created_at: Mapped[datetime] = mapped_column(  # pyright: ignore
+        DateTimeUTC(timezone=True),
+        default=datetime.now(timezone.utc),
+    )
     """Date/time of instance creation."""
-    updated: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore
+    updated_at: Mapped[datetime] = mapped_column(  # pyright: ignore
+        DateTimeUTC(timezone=True),
+        default=datetime.now(timezone.utc),
+    )
     """Date/time of instance last update."""
 
 
@@ -152,7 +158,7 @@
             AnyUrl: String,
             AnyHttpUrl: String,
             dict: JsonB,
-            datetime: DateTime,
+            datetime: DateTimeUTC,
             date: Date,
         },
     )
diff --git a/litestar/contrib/sqlalchemy/types.py b/litestar/contrib/sqlalchemy/types.py
--- a/litestar/contrib/sqlalchemy/types.py
+++ b/litestar/contrib/sqlalchemy/types.py
@@ -1,10 +1,11 @@
 from __future__ import annotations
 
+import datetime
 import uuid
 from base64 import b64decode
 from typing import TYPE_CHECKING, Any, cast
 
-from sqlalchemy import text, util
+from sqlalchemy import DateTime, text, util
 from sqlalchemy.dialects.oracle import BLOB as ORA_BLOB
 from sqlalchemy.dialects.oracle import RAW as ORA_RAW
 from sqlalchemy.dialects.postgresql import JSONB as PG_JSONB
@@ -145,6 +146,30 @@
         table.append_constraint(e)
 
 
+class DateTimeUTC(TypeDecorator):
+    """Timezone Aware DateTime.
+
+    Ensure UTC is stored in the database and that TZ aware dates are returned for all dialects.
+    """
+
+    impl = DateTime(timezone=True)
+    cache_ok = True
+
+    def process_bind_param(self, value: datetime.datetime | None, dialect: Dialect) -> datetime.datetime | None:
+        if value is None:
+            return value
+        if not value.tzinfo:
+            raise TypeError("tzinfo is required")
+        return value.astimezone(datetime.timezone.utc)
+
+    def process_result_value(self, value: datetime.datetime | None, dialect: Dialect) -> datetime.datetime | None:
+        if value is None:
+            return value
+        if value.tzinfo is None:
+            return value.replace(tzinfo=datetime.timezone.utc)
+        return value
+
+
 BigIntIdentity = BigInteger().with_variant(Integer, "sqlite")
 """A ``BigInteger`` variant that reverts to an ``Integer`` for unsupported variants."""

{"golden_diff": "diff --git a/litestar/contrib/sqlalchemy/base.py b/litestar/contrib/sqlalchemy/base.py\n--- a/litestar/contrib/sqlalchemy/base.py\n+++ b/litestar/contrib/sqlalchemy/base.py\n@@ -2,12 +2,12 @@\n from __future__ import annotations\n \n import re\n-from datetime import date, datetime\n+from datetime import date, datetime, timezone\n from typing import TYPE_CHECKING, Any, ClassVar, Protocol, TypeVar, runtime_checkable\n from uuid import UUID, uuid4\n \n from pydantic import AnyHttpUrl, AnyUrl, EmailStr\n-from sqlalchemy import Date, DateTime, MetaData, Sequence, String\n+from sqlalchemy import Date, MetaData, Sequence, String\n from sqlalchemy.event import listens_for\n from sqlalchemy.orm import (\n     DeclarativeBase,\n@@ -19,7 +19,7 @@\n     registry,\n )\n \n-from .types import GUID, BigIntIdentity, JsonB\n+from .types import GUID, BigIntIdentity, DateTimeUTC, JsonB\n \n if TYPE_CHECKING:\n     from sqlalchemy.sql import FromClause\n@@ -65,8 +65,8 @@\n             session.\n     \"\"\"\n     for instance in session.dirty:\n-        if hasattr(instance, \"updated\"):\n-            instance.updated = datetime.now()  # noqa: DTZ005\n+        if hasattr(instance, \"updated_at\"):\n+            instance.updated = (datetime.now(timezone.utc),)\n \n \n @runtime_checkable\n@@ -112,9 +112,15 @@\n class AuditColumns:\n     \"\"\"Created/Updated At Fields Mixin.\"\"\"\n \n-    created: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore\n+    created_at: Mapped[datetime] = mapped_column(  # pyright: ignore\n+        DateTimeUTC(timezone=True),\n+        default=datetime.now(timezone.utc),\n+    )\n     \"\"\"Date/time of instance creation.\"\"\"\n-    updated: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore\n+    updated_at: Mapped[datetime] = mapped_column(  # pyright: ignore\n+        DateTimeUTC(timezone=True),\n+        default=datetime.now(timezone.utc),\n+    )\n     \"\"\"Date/time of instance last update.\"\"\"\n \n \n@@ -152,7 +158,7 @@\n             AnyUrl: String,\n             AnyHttpUrl: String,\n             dict: JsonB,\n-            datetime: DateTime,\n+            datetime: DateTimeUTC,\n             date: Date,\n         },\n     )\ndiff --git a/litestar/contrib/sqlalchemy/types.py b/litestar/contrib/sqlalchemy/types.py\n--- a/litestar/contrib/sqlalchemy/types.py\n+++ b/litestar/contrib/sqlalchemy/types.py\n@@ -1,10 +1,11 @@\n from __future__ import annotations\n \n+import datetime\n import uuid\n from base64 import b64decode\n from typing import TYPE_CHECKING, Any, cast\n \n-from sqlalchemy import text, util\n+from sqlalchemy import DateTime, text, util\n from sqlalchemy.dialects.oracle import BLOB as ORA_BLOB\n from sqlalchemy.dialects.oracle import RAW as ORA_RAW\n from sqlalchemy.dialects.postgresql import JSONB as PG_JSONB\n@@ -145,6 +146,30 @@\n         table.append_constraint(e)\n \n \n+class DateTimeUTC(TypeDecorator):\n+    \"\"\"Timezone Aware DateTime.\n+\n+    Ensure UTC is stored in the database and that TZ aware dates are returned for all dialects.\n+    \"\"\"\n+\n+    impl = DateTime(timezone=True)\n+    cache_ok = True\n+\n+    def process_bind_param(self, value: datetime.datetime | None, dialect: Dialect) -> datetime.datetime | None:\n+        if value is None:\n+            return value\n+        if not value.tzinfo:\n+            raise TypeError(\"tzinfo is required\")\n+        return value.astimezone(datetime.timezone.utc)\n+\n+    def process_result_value(self, value: datetime.datetime | None, dialect: Dialect) -> datetime.datetime | None:\n+        if value is None:\n+            return value\n+        if value.tzinfo is None:\n+            return value.replace(tzinfo=datetime.timezone.utc)\n+        return value\n+\n+\n BigIntIdentity = BigInteger().with_variant(Integer, \"sqlite\")\n \"\"\"A ``BigInteger`` variant that reverts to an ``Integer`` for unsupported variants.\"\"\"\n", "issue": "StaticFilesConfig and virtual directories\nI'm trying to write a ``FileSystemProtocol`` to load files from the package data using [importlib_resources](https://importlib-resources.readthedocs.io/en/latest/using.html#). But because ``directories`` is defined as ``DirectoryPath``, pydantic checks if the given directories exist in the local filesystem. \r\n\r\nThis is not generally true, especially in any kind of virtual filesystem (e.g. a zipped package). I think this condition should be relaxed to support virtual filesystems.\r\n\r\nhttps://github.com/starlite-api/starlite/blob/9bb6dcd57c10a591377cf8e3a537e9292566d5b9/starlite/config/static_files.py#L32\n", "before_files": [{"content": "\"\"\"Application ORM configuration.\"\"\"\nfrom __future__ import annotations\n\nimport re\nfrom datetime import date, datetime\nfrom typing import TYPE_CHECKING, Any, ClassVar, Protocol, TypeVar, runtime_checkable\nfrom uuid import UUID, uuid4\n\nfrom pydantic import AnyHttpUrl, AnyUrl, EmailStr\nfrom sqlalchemy import Date, DateTime, MetaData, Sequence, String\nfrom sqlalchemy.event import listens_for\nfrom sqlalchemy.orm import (\n    DeclarativeBase,\n    Mapped,\n    Session,\n    declared_attr,\n    mapped_column,\n    orm_insert_sentinel,\n    registry,\n)\n\nfrom .types import GUID, BigIntIdentity, JsonB\n\nif TYPE_CHECKING:\n    from sqlalchemy.sql import FromClause\n\n__all__ = (\n    \"AuditColumns\",\n    \"BigIntAuditBase\",\n    \"BigIntBase\",\n    \"BigIntPrimaryKey\",\n    \"CommonTableAttributes\",\n    \"create_registry\",\n    \"ModelProtocol\",\n    \"touch_updated_timestamp\",\n    \"UUIDAuditBase\",\n    \"UUIDBase\",\n    \"UUIDPrimaryKey\",\n)\n\n\nUUIDBaseT = TypeVar(\"UUIDBaseT\", bound=\"UUIDBase\")\nBigIntBaseT = TypeVar(\"BigIntBaseT\", bound=\"BigIntBase\")\n\nconvention = {\n    \"ix\": \"ix_%(column_0_label)s\",\n    \"uq\": \"uq_%(table_name)s_%(column_0_name)s\",\n    \"ck\": \"ck_%(table_name)s_%(constraint_name)s\",\n    \"fk\": \"fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s\",\n    \"pk\": \"pk_%(table_name)s\",\n}\n\"\"\"Templates for automated constraint name generation.\"\"\"\n\n\n@listens_for(Session, \"before_flush\")\ndef touch_updated_timestamp(session: Session, *_: Any) -> None:\n    \"\"\"Set timestamp on update.\n\n    Called from SQLAlchemy's\n    :meth:`before_flush <sqlalchemy.orm.SessionEvents.before_flush>` event to bump the ``updated``\n    timestamp on modified instances.\n\n    Args:\n        session: The sync :class:`Session <sqlalchemy.orm.Session>` instance that underlies the async\n            session.\n    \"\"\"\n    for instance in session.dirty:\n        if hasattr(instance, \"updated\"):\n            instance.updated = datetime.now()  # noqa: DTZ005\n\n\n@runtime_checkable\nclass ModelProtocol(Protocol):\n    \"\"\"The base SQLAlchemy model protocol.\"\"\"\n\n    __table__: FromClause\n    __name__: ClassVar[str]\n\n    def to_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:\n        \"\"\"Convert model to dictionary.\n\n        Returns:\n            dict[str, Any]: A dict representation of the model\n        \"\"\"\n        ...\n\n\nclass UUIDPrimaryKey:\n    \"\"\"UUID Primary Key Field Mixin.\"\"\"\n\n    id: Mapped[UUID] = mapped_column(default=uuid4, primary_key=True)  # pyright: ignore\n    \"\"\"UUID Primary key column.\"\"\"\n\n    @declared_attr\n    def _sentinel(cls) -> Mapped[int]:\n        return orm_insert_sentinel()\n\n\nclass BigIntPrimaryKey:\n    \"\"\"BigInt Primary Key Field Mixin.\"\"\"\n\n    @declared_attr\n    def id(cls) -> Mapped[int]:\n        \"\"\"BigInt Primary key column.\"\"\"\n        return mapped_column(\n            BigIntIdentity,\n            Sequence(f\"{cls.__tablename__}_id_seq\", optional=False),  # type: ignore[attr-defined] # pyright: ignore\n            primary_key=True,\n        )\n\n\nclass AuditColumns:\n    \"\"\"Created/Updated At Fields Mixin.\"\"\"\n\n    created: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore\n    \"\"\"Date/time of instance creation.\"\"\"\n    updated: Mapped[datetime] = mapped_column(default=datetime.now)  # pyright: ignore\n    \"\"\"Date/time of instance last update.\"\"\"\n\n\nclass CommonTableAttributes:\n    \"\"\"Common attributes for SQLALchemy tables.\"\"\"\n\n    __name__: ClassVar[str]\n    __table__: FromClause\n\n    # noinspection PyMethodParameters\n    @declared_attr.directive\n    def __tablename__(cls) -> str:  # pylint: disable=no-self-argument\n        \"\"\"Infer table name from class name.\"\"\"\n        regexp = re.compile(\"((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))\")\n        return regexp.sub(r\"_\\1\", cls.__name__).lower()\n\n    def to_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:\n        \"\"\"Convert model to dictionary.\n\n        Returns:\n            dict[str, Any]: A dict representation of the model\n        \"\"\"\n        exclude = {\"_sentinel\"}.union(self._sa_instance_state.unloaded).union(exclude or [])  # type: ignore[attr-defined]\n        return {field.name: getattr(self, field.name) for field in self.__table__.columns if field.name not in exclude}\n\n\ndef create_registry() -> registry:\n    \"\"\"Create a new SQLAlchemy registry.\"\"\"\n    meta = MetaData(naming_convention=convention)\n    return registry(\n        metadata=meta,\n        type_annotation_map={\n            UUID: GUID,\n            EmailStr: String,\n            AnyUrl: String,\n            AnyHttpUrl: String,\n            dict: JsonB,\n            datetime: DateTime,\n            date: Date,\n        },\n    )\n\n\norm_registry = create_registry()\n\n\nclass UUIDBase(UUIDPrimaryKey, CommonTableAttributes, DeclarativeBase):\n    \"\"\"Base for all SQLAlchemy declarative models with UUID primary keys.\"\"\"\n\n    registry = orm_registry\n\n\nclass UUIDAuditBase(CommonTableAttributes, UUIDPrimaryKey, AuditColumns, DeclarativeBase):\n    \"\"\"Base for declarative models with UUID primary keys and audit columns.\"\"\"\n\n    registry = orm_registry\n\n\nclass BigIntBase(BigIntPrimaryKey, CommonTableAttributes, DeclarativeBase):\n    \"\"\"Base for all SQLAlchemy declarative models with BigInt primary keys.\"\"\"\n\n    registry = orm_registry\n\n\nclass BigIntAuditBase(CommonTableAttributes, BigIntPrimaryKey, AuditColumns, DeclarativeBase):\n    \"\"\"Base for declarative models with BigInt primary keys and audit columns.\"\"\"\n\n    registry = orm_registry\n", "path": "litestar/contrib/sqlalchemy/base.py"}, {"content": "from __future__ import annotations\n\nimport uuid\nfrom base64 import b64decode\nfrom typing import TYPE_CHECKING, Any, cast\n\nfrom sqlalchemy import text, util\nfrom sqlalchemy.dialects.oracle import BLOB as ORA_BLOB\nfrom sqlalchemy.dialects.oracle import RAW as ORA_RAW\nfrom sqlalchemy.dialects.postgresql import JSONB as PG_JSONB\nfrom sqlalchemy.dialects.postgresql import UUID as PG_UUID\nfrom sqlalchemy.types import BINARY, CHAR, BigInteger, Integer, SchemaType, TypeDecorator, TypeEngine\nfrom sqlalchemy.types import JSON as _JSON\n\nfrom litestar.serialization import decode_json, encode_json\n\nif TYPE_CHECKING:\n    from sqlalchemy.engine import Dialect\n\n\nclass GUID(TypeDecorator):\n    \"\"\"Platform-independent GUID type.\n\n    Uses PostgreSQL's UUID type, Oracle's RAW(16) type, otherwise uses\n    BINARY(16) or CHAR(32), storing as stringified hex values.\n\n    Will accept stringified UUIDs as a hexstring or an actual UUID\n\n    \"\"\"\n\n    impl = BINARY(16)\n    cache_ok = True\n\n    @property\n    def python_type(self) -> type[uuid.UUID]:\n        return uuid.UUID\n\n    def __init__(self, *args: Any, binary: bool = True, **kwargs: Any) -> None:\n        self.binary = binary\n\n    def load_dialect_impl(self, dialect: Dialect) -> Any:\n        if dialect.name in {\"postgresql\", \"duckdb\"}:\n            return dialect.type_descriptor(PG_UUID())\n        if dialect.name == \"oracle\":\n            return dialect.type_descriptor(ORA_RAW(16))\n        if self.binary:\n            return dialect.type_descriptor(BINARY(16))\n        return dialect.type_descriptor(CHAR(32))\n\n    def process_bind_param(self, value: bytes | str | uuid.UUID | None, dialect: Dialect) -> bytes | str | None:\n        if value is None:\n            return value\n        if dialect.name in {\"postgresql\", \"duckdb\"}:\n            return str(value)\n        value = self.to_uuid(value)\n        if value is None:\n            return value\n        if dialect.name in {\"oracle\", \"spanner+spanner\"}:\n            return value.bytes\n        return value.bytes if self.binary else value.hex\n\n    def process_result_value(self, value: bytes | str | uuid.UUID | None, dialect: Dialect) -> uuid.UUID | None:\n        if value is None:\n            return value\n        if isinstance(value, uuid.UUID):\n            return value\n        if dialect.name == \"spanner+spanner\":\n            return uuid.UUID(bytes=b64decode(value))\n        if self.binary:\n            return uuid.UUID(bytes=cast(\"bytes\", value))\n        return uuid.UUID(hex=cast(\"str\", value))\n\n    @staticmethod\n    def to_uuid(value: Any) -> uuid.UUID | None:\n        if isinstance(value, uuid.UUID) or value is None:\n            return value\n        try:\n            value = uuid.UUID(hex=value)\n        except (TypeError, ValueError):\n            value = uuid.UUID(bytes=value)\n        return cast(\"uuid.UUID | None\", value)\n\n\nclass ORA_JSONB(TypeDecorator, SchemaType):  # type: ignore  # noqa: N801\n    \"\"\"Oracle Binary JSON type.\n\n    JsonB = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_JSONB, \"oracle\")\n\n    \"\"\"\n\n    impl = ORA_BLOB\n    cache_ok = True\n\n    @property\n    def python_type(self) -> type[dict]:\n        return dict\n\n    def __init__(self, *args: Any, **kwargs: Any) -> None:\n        \"\"\"Initialize JSON type\"\"\"\n        self.name = kwargs.pop(\"name\", None)\n        self.oracle_strict = kwargs.pop(\"oracle_strict\", True)\n\n    def coerce_compared_value(self, op: Any, value: Any) -> Any:\n        return self.impl.coerce_compared_value(op=op, value=value)  # type: ignore\n\n    def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:\n        return dialect.type_descriptor(ORA_BLOB())\n\n    def process_bind_param(self, value: Any, dialect: Dialect) -> Any | None:\n        if value is None:\n            return value\n        return encode_json(value)\n\n    def process_result_value(self, value: bytes | None, dialect: Dialect) -> Any | None:\n        if value is None:\n            return value\n        return decode_json(value)\n\n    def _should_create_constraint(self, compiler: Any, **kw: Any) -> bool:\n        return bool(compiler.dialect.name == \"oracle\")\n\n    def _variant_mapping_for_set_table(self, column: Any) -> dict | None:\n        if column.type._variant_mapping:\n            variant_mapping = dict(column.type._variant_mapping)\n            variant_mapping[\"_default\"] = column.type\n        else:\n            variant_mapping = None\n        return variant_mapping\n\n    @util.preload_module(\"sqlalchemy.sql.schema\")\n    def _set_table(self, column: Any, table: Any) -> None:\n        schema = util.preloaded.sql_schema\n        variant_mapping = self._variant_mapping_for_set_table(column)\n        constraint_options = \"(strict)\" if self.oracle_strict else \"\"\n        sqltext = text(f\"{column.name} is json {constraint_options}\")\n        e = schema.CheckConstraint(\n            sqltext,\n            name=f\"{column.name}_is_json\",\n            _create_rule=util.portable_instancemethod(  # type: ignore[no-untyped-call]\n                self._should_create_constraint,\n                {\"variant_mapping\": variant_mapping},\n            ),\n            _type_bound=True,\n        )\n        table.append_constraint(e)\n\n\nBigIntIdentity = BigInteger().with_variant(Integer, \"sqlite\")\n\"\"\"A ``BigInteger`` variant that reverts to an ``Integer`` for unsupported variants.\"\"\"\n\nJsonB = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_JSONB, \"oracle\")\n\"\"\"A JSON type that uses  native ``JSONB`` where possible and ``Binary`` or ``Blob`` as\nan alternative.\n\"\"\"\n", "path": "litestar/contrib/sqlalchemy/types.py"}], "after_files": [{"content": "\"\"\"Application ORM configuration.\"\"\"\nfrom __future__ import annotations\n\nimport re\nfrom datetime import date, datetime, timezone\nfrom typing import TYPE_CHECKING, Any, ClassVar, Protocol, TypeVar, runtime_checkable\nfrom uuid import UUID, uuid4\n\nfrom pydantic import AnyHttpUrl, AnyUrl, EmailStr\nfrom sqlalchemy import Date, MetaData, Sequence, String\nfrom sqlalchemy.event import listens_for\nfrom sqlalchemy.orm import (\n    DeclarativeBase,\n    Mapped,\n    Session,\n    declared_attr,\n    mapped_column,\n    orm_insert_sentinel,\n    registry,\n)\n\nfrom .types import GUID, BigIntIdentity, DateTimeUTC, JsonB\n\nif TYPE_CHECKING:\n    from sqlalchemy.sql import FromClause\n\n__all__ = (\n    \"AuditColumns\",\n    \"BigIntAuditBase\",\n    \"BigIntBase\",\n    \"BigIntPrimaryKey\",\n    \"CommonTableAttributes\",\n    \"create_registry\",\n    \"ModelProtocol\",\n    \"touch_updated_timestamp\",\n    \"UUIDAuditBase\",\n    \"UUIDBase\",\n    \"UUIDPrimaryKey\",\n)\n\n\nUUIDBaseT = TypeVar(\"UUIDBaseT\", bound=\"UUIDBase\")\nBigIntBaseT = TypeVar(\"BigIntBaseT\", bound=\"BigIntBase\")\n\nconvention = {\n    \"ix\": \"ix_%(column_0_label)s\",\n    \"uq\": \"uq_%(table_name)s_%(column_0_name)s\",\n    \"ck\": \"ck_%(table_name)s_%(constraint_name)s\",\n    \"fk\": \"fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s\",\n    \"pk\": \"pk_%(table_name)s\",\n}\n\"\"\"Templates for automated constraint name generation.\"\"\"\n\n\n@listens_for(Session, \"before_flush\")\ndef touch_updated_timestamp(session: Session, *_: Any) -> None:\n    \"\"\"Set timestamp on update.\n\n    Called from SQLAlchemy's\n    :meth:`before_flush <sqlalchemy.orm.SessionEvents.before_flush>` event to bump the ``updated``\n    timestamp on modified instances.\n\n    Args:\n        session: The sync :class:`Session <sqlalchemy.orm.Session>` instance that underlies the async\n            session.\n    \"\"\"\n    for instance in session.dirty:\n        if hasattr(instance, \"updated_at\"):\n            instance.updated = (datetime.now(timezone.utc),)\n\n\n@runtime_checkable\nclass ModelProtocol(Protocol):\n    \"\"\"The base SQLAlchemy model protocol.\"\"\"\n\n    __table__: FromClause\n    __name__: ClassVar[str]\n\n    def to_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:\n        \"\"\"Convert model to dictionary.\n\n        Returns:\n            dict[str, Any]: A dict representation of the model\n        \"\"\"\n        ...\n\n\nclass UUIDPrimaryKey:\n    \"\"\"UUID Primary Key Field Mixin.\"\"\"\n\n    id: Mapped[UUID] = mapped_column(default=uuid4, primary_key=True)  # pyright: ignore\n    \"\"\"UUID Primary key column.\"\"\"\n\n    @declared_attr\n    def _sentinel(cls) -> Mapped[int]:\n        return orm_insert_sentinel()\n\n\nclass BigIntPrimaryKey:\n    \"\"\"BigInt Primary Key Field Mixin.\"\"\"\n\n    @declared_attr\n    def id(cls) -> Mapped[int]:\n        \"\"\"BigInt Primary key column.\"\"\"\n        return mapped_column(\n            BigIntIdentity,\n            Sequence(f\"{cls.__tablename__}_id_seq\", optional=False),  # type: ignore[attr-defined] # pyright: ignore\n            primary_key=True,\n        )\n\n\nclass AuditColumns:\n    \"\"\"Created/Updated At Fields Mixin.\"\"\"\n\n    created_at: Mapped[datetime] = mapped_column(  # pyright: ignore\n        DateTimeUTC(timezone=True),\n        default=datetime.now(timezone.utc),\n    )\n    \"\"\"Date/time of instance creation.\"\"\"\n    updated_at: Mapped[datetime] = mapped_column(  # pyright: ignore\n        DateTimeUTC(timezone=True),\n        default=datetime.now(timezone.utc),\n    )\n    \"\"\"Date/time of instance last update.\"\"\"\n\n\nclass CommonTableAttributes:\n    \"\"\"Common attributes for SQLALchemy tables.\"\"\"\n\n    __name__: ClassVar[str]\n    __table__: FromClause\n\n    # noinspection PyMethodParameters\n    @declared_attr.directive\n    def __tablename__(cls) -> str:  # pylint: disable=no-self-argument\n        \"\"\"Infer table name from class name.\"\"\"\n        regexp = re.compile(\"((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))\")\n        return regexp.sub(r\"_\\1\", cls.__name__).lower()\n\n    def to_dict(self, exclude: set[str] | None = None) -> dict[str, Any]:\n        \"\"\"Convert model to dictionary.\n\n        Returns:\n            dict[str, Any]: A dict representation of the model\n        \"\"\"\n        exclude = {\"_sentinel\"}.union(self._sa_instance_state.unloaded).union(exclude or [])  # type: ignore[attr-defined]\n        return {field.name: getattr(self, field.name) for field in self.__table__.columns if field.name not in exclude}\n\n\ndef create_registry() -> registry:\n    \"\"\"Create a new SQLAlchemy registry.\"\"\"\n    meta = MetaData(naming_convention=convention)\n    return registry(\n        metadata=meta,\n        type_annotation_map={\n            UUID: GUID,\n            EmailStr: String,\n            AnyUrl: String,\n            AnyHttpUrl: String,\n            dict: JsonB,\n            datetime: DateTimeUTC,\n            date: Date,\n        },\n    )\n\n\norm_registry = create_registry()\n\n\nclass UUIDBase(UUIDPrimaryKey, CommonTableAttributes, DeclarativeBase):\n    \"\"\"Base for all SQLAlchemy declarative models with UUID primary keys.\"\"\"\n\n    registry = orm_registry\n\n\nclass UUIDAuditBase(CommonTableAttributes, UUIDPrimaryKey, AuditColumns, DeclarativeBase):\n    \"\"\"Base for declarative models with UUID primary keys and audit columns.\"\"\"\n\n    registry = orm_registry\n\n\nclass BigIntBase(BigIntPrimaryKey, CommonTableAttributes, DeclarativeBase):\n    \"\"\"Base for all SQLAlchemy declarative models with BigInt primary keys.\"\"\"\n\n    registry = orm_registry\n\n\nclass BigIntAuditBase(CommonTableAttributes, BigIntPrimaryKey, AuditColumns, DeclarativeBase):\n    \"\"\"Base for declarative models with BigInt primary keys and audit columns.\"\"\"\n\n    registry = orm_registry\n", "path": "litestar/contrib/sqlalchemy/base.py"}, {"content": "from __future__ import annotations\n\nimport datetime\nimport uuid\nfrom base64 import b64decode\nfrom typing import TYPE_CHECKING, Any, cast\n\nfrom sqlalchemy import DateTime, text, util\nfrom sqlalchemy.dialects.oracle import BLOB as ORA_BLOB\nfrom sqlalchemy.dialects.oracle import RAW as ORA_RAW\nfrom sqlalchemy.dialects.postgresql import JSONB as PG_JSONB\nfrom sqlalchemy.dialects.postgresql import UUID as PG_UUID\nfrom sqlalchemy.types import BINARY, CHAR, BigInteger, Integer, SchemaType, TypeDecorator, TypeEngine\nfrom sqlalchemy.types import JSON as _JSON\n\nfrom litestar.serialization import decode_json, encode_json\n\nif TYPE_CHECKING:\n    from sqlalchemy.engine import Dialect\n\n\nclass GUID(TypeDecorator):\n    \"\"\"Platform-independent GUID type.\n\n    Uses PostgreSQL's UUID type, Oracle's RAW(16) type, otherwise uses\n    BINARY(16) or CHAR(32), storing as stringified hex values.\n\n    Will accept stringified UUIDs as a hexstring or an actual UUID\n\n    \"\"\"\n\n    impl = BINARY(16)\n    cache_ok = True\n\n    @property\n    def python_type(self) -> type[uuid.UUID]:\n        return uuid.UUID\n\n    def __init__(self, *args: Any, binary: bool = True, **kwargs: Any) -> None:\n        self.binary = binary\n\n    def load_dialect_impl(self, dialect: Dialect) -> Any:\n        if dialect.name in {\"postgresql\", \"duckdb\"}:\n            return dialect.type_descriptor(PG_UUID())\n        if dialect.name == \"oracle\":\n            return dialect.type_descriptor(ORA_RAW(16))\n        if self.binary:\n            return dialect.type_descriptor(BINARY(16))\n        return dialect.type_descriptor(CHAR(32))\n\n    def process_bind_param(self, value: bytes | str | uuid.UUID | None, dialect: Dialect) -> bytes | str | None:\n        if value is None:\n            return value\n        if dialect.name in {\"postgresql\", \"duckdb\"}:\n            return str(value)\n        value = self.to_uuid(value)\n        if value is None:\n            return value\n        if dialect.name in {\"oracle\", \"spanner+spanner\"}:\n            return value.bytes\n        return value.bytes if self.binary else value.hex\n\n    def process_result_value(self, value: bytes | str | uuid.UUID | None, dialect: Dialect) -> uuid.UUID | None:\n        if value is None:\n            return value\n        if isinstance(value, uuid.UUID):\n            return value\n        if dialect.name == \"spanner+spanner\":\n            return uuid.UUID(bytes=b64decode(value))\n        if self.binary:\n            return uuid.UUID(bytes=cast(\"bytes\", value))\n        return uuid.UUID(hex=cast(\"str\", value))\n\n    @staticmethod\n    def to_uuid(value: Any) -> uuid.UUID | None:\n        if isinstance(value, uuid.UUID) or value is None:\n            return value\n        try:\n            value = uuid.UUID(hex=value)\n        except (TypeError, ValueError):\n            value = uuid.UUID(bytes=value)\n        return cast(\"uuid.UUID | None\", value)\n\n\nclass ORA_JSONB(TypeDecorator, SchemaType):  # type: ignore  # noqa: N801\n    \"\"\"Oracle Binary JSON type.\n\n    JsonB = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_JSONB, \"oracle\")\n\n    \"\"\"\n\n    impl = ORA_BLOB\n    cache_ok = True\n\n    @property\n    def python_type(self) -> type[dict]:\n        return dict\n\n    def __init__(self, *args: Any, **kwargs: Any) -> None:\n        \"\"\"Initialize JSON type\"\"\"\n        self.name = kwargs.pop(\"name\", None)\n        self.oracle_strict = kwargs.pop(\"oracle_strict\", True)\n\n    def coerce_compared_value(self, op: Any, value: Any) -> Any:\n        return self.impl.coerce_compared_value(op=op, value=value)  # type: ignore\n\n    def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:\n        return dialect.type_descriptor(ORA_BLOB())\n\n    def process_bind_param(self, value: Any, dialect: Dialect) -> Any | None:\n        if value is None:\n            return value\n        return encode_json(value)\n\n    def process_result_value(self, value: bytes | None, dialect: Dialect) -> Any | None:\n        if value is None:\n            return value\n        return decode_json(value)\n\n    def _should_create_constraint(self, compiler: Any, **kw: Any) -> bool:\n        return bool(compiler.dialect.name == \"oracle\")\n\n    def _variant_mapping_for_set_table(self, column: Any) -> dict | None:\n        if column.type._variant_mapping:\n            variant_mapping = dict(column.type._variant_mapping)\n            variant_mapping[\"_default\"] = column.type\n        else:\n            variant_mapping = None\n        return variant_mapping\n\n    @util.preload_module(\"sqlalchemy.sql.schema\")\n    def _set_table(self, column: Any, table: Any) -> None:\n        schema = util.preloaded.sql_schema\n        variant_mapping = self._variant_mapping_for_set_table(column)\n        constraint_options = \"(strict)\" if self.oracle_strict else \"\"\n        sqltext = text(f\"{column.name} is json {constraint_options}\")\n        e = schema.CheckConstraint(\n            sqltext,\n            name=f\"{column.name}_is_json\",\n            _create_rule=util.portable_instancemethod(  # type: ignore[no-untyped-call]\n                self._should_create_constraint,\n                {\"variant_mapping\": variant_mapping},\n            ),\n            _type_bound=True,\n        )\n        table.append_constraint(e)\n\n\nclass DateTimeUTC(TypeDecorator):\n    \"\"\"Timezone Aware DateTime.\n\n    Ensure UTC is stored in the database and that TZ aware dates are returned for all dialects.\n    \"\"\"\n\n    impl = DateTime(timezone=True)\n    cache_ok = True\n\n    def process_bind_param(self, value: datetime.datetime | None, dialect: Dialect) -> datetime.datetime | None:\n        if value is None:\n            return value\n        if not value.tzinfo:\n            raise TypeError(\"tzinfo is required\")\n        return value.astimezone(datetime.timezone.utc)\n\n    def process_result_value(self, value: datetime.datetime | None, dialect: Dialect) -> datetime.datetime | None:\n        if value is None:\n            return value\n        if value.tzinfo is None:\n            return value.replace(tzinfo=datetime.timezone.utc)\n        return value\n\n\nBigIntIdentity = BigInteger().with_variant(Integer, \"sqlite\")\n\"\"\"A ``BigInteger`` variant that reverts to an ``Integer`` for unsupported variants.\"\"\"\n\nJsonB = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_JSONB, \"oracle\")\n\"\"\"A JSON type that uses  native ``JSONB`` where possible and ``Binary`` or ``Blob`` as\nan alternative.\n\"\"\"\n", "path": "litestar/contrib/sqlalchemy/types.py"}]}