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, JSON, BigIntIdentity
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 = exclude.union("_sentinel") if exclude else {"_sentinel"}
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: JSON,
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
13 from sqlalchemy.types import JSON as _JSON
14 
15 if TYPE_CHECKING:
16     from sqlalchemy.engine import Dialect
17 
18 BigIntIdentity = BigInteger().with_variant(Integer, "sqlite")
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 JSON(TypeDecorator, SchemaType):  # type: ignore
85     """Platform-independent JSON type.
86 
87     Uses JSONB type for postgres, BLOB for Oracle, otherwise uses the generic JSON data type.
88 
89     JSON = _JSON().with_variant(PG_JSONB, "postgresql").with_variant(ORA_BLOB, "oracle")
90 
91     """
92 
93     impl = _JSON
94     cache_ok = True
95 
96     @property
97     def python_type(self) -> type[dict]:
98         return dict
99 
100     def __init__(self, *args: Any, **kwargs: Any) -> None:
101         """Initialize JSON type"""
102         self.name = kwargs.pop("name", None)
103         self.oracle_strict = kwargs.pop("oracle_strict", True)
104 
105     def load_dialect_impl(self, dialect: Dialect) -> Any:
106         if dialect.name == "postgresql":
107             return dialect.type_descriptor(PG_JSONB())  # type: ignore
108         if dialect.name == "oracle":
109             return dialect.type_descriptor(ORA_BLOB())
110         return dialect.type_descriptor(_JSON())
111 
112     def _should_create_constraint(self, compiler: Any, **kw: Any) -> bool:
113         return bool(compiler.dialect.name == "oracle")
114 
115     def _variant_mapping_for_set_table(self, column: Any) -> dict | None:
116         if column.type._variant_mapping:
117             variant_mapping = dict(column.type._variant_mapping)
118             variant_mapping["_default"] = column.type
119         else:
120             variant_mapping = None
121         return variant_mapping
122 
123     @util.preload_module("sqlalchemy.sql.schema")
124     def _set_table(self, column: Any, table: Any) -> None:
125         schema = util.preloaded.sql_schema
126         variant_mapping = self._variant_mapping_for_set_table(column)
127         constraint_options = "(strict)" if self.oracle_strict else ""
128         sqltext = text(f"{column.name} is json {constraint_options}")
129         e = schema.CheckConstraint(
130             sqltext,
131             name=f"{column.name}_is_json",
132             _create_rule=util.portable_instancemethod(  # type: ignore[no-untyped-call]
133                 self._should_create_constraint,
134                 {"variant_mapping": variant_mapping},
135             ),
136             _type_bound=True,
137         )
138         table.append_constraint(e)
139
```
--- 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
@@ -19,7 +19,7 @@
     registry,
 )
 
-from .types import GUID, JSON, BigIntIdentity
+from .types import GUID, BigIntIdentity, JsonB
 
 if TYPE_CHECKING:
     from sqlalchemy.sql import FromClause
@@ -151,7 +151,7 @@
             EmailStr: String,
             AnyUrl: String,
             AnyHttpUrl: String,
-            dict: JSON,
+            dict: JsonB,
             datetime: DateTime,
             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
@@ -9,14 +9,14 @@
 from sqlalchemy.dialects.oracle import RAW as ORA_RAW
 from sqlalchemy.dialects.postgresql import JSONB as PG_JSONB
 from sqlalchemy.dialects.postgresql import UUID as PG_UUID
-from sqlalchemy.types import BINARY, CHAR, BigInteger, Integer, SchemaType, TypeDecorator
+from sqlalchemy.types import BINARY, CHAR, BigInteger, Integer, SchemaType, TypeDecorator, TypeEngine
 from sqlalchemy.types import JSON as _JSON
 
+from litestar.serialization import decode_json, encode_json
+
 if TYPE_CHECKING:
     from sqlalchemy.engine import Dialect
 
-BigIntIdentity = BigInteger().with_variant(Integer, "sqlite")
-
 
 class GUID(TypeDecorator):
     """Platform-independent GUID type.
@@ -81,16 +81,14 @@
         return cast("uuid.UUID | None", value)
 
 
-class JSON(TypeDecorator, SchemaType):  # type: ignore
-    """Platform-independent JSON type.
-
-    Uses JSONB type for postgres, BLOB for Oracle, otherwise uses the generic JSON data type.
+class ORA_JSONB(TypeDecorator, SchemaType):  # type: ignore  # noqa: N801
+    """Oracle Binary JSON type.
 
-    JSON = _JSON().with_variant(PG_JSONB, "postgresql").with_variant(ORA_BLOB, "oracle")
+    JsonB = _JSON().with_variant(PG_JSONB, "postgresql").with_variant(ORA_JSONB, "oracle")
 
     """
 
-    impl = _JSON
+    impl = ORA_BLOB
     cache_ok = True
 
     @property
@@ -102,12 +100,21 @@
         self.name = kwargs.pop("name", None)
         self.oracle_strict = kwargs.pop("oracle_strict", True)
 
-    def load_dialect_impl(self, dialect: Dialect) -> Any:
-        if dialect.name == "postgresql":
-            return dialect.type_descriptor(PG_JSONB())  # type: ignore
-        if dialect.name == "oracle":
-            return dialect.type_descriptor(ORA_BLOB())
-        return dialect.type_descriptor(_JSON())
+    def coerce_compared_value(self, op: Any, value: Any) -> Any:
+        return self.impl.coerce_compared_value(op=op, value=value)  # type: ignore
+
+    def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:
+        return dialect.type_descriptor(ORA_BLOB())
+
+    def process_bind_param(self, value: Any, dialect: Dialect) -> Any | None:
+        if value is None:
+            return value
+        return encode_json(value)
+
+    def process_result_value(self, value: bytes | None, dialect: Dialect) -> Any | None:
+        if value is None:
+            return value
+        return decode_json(value)
 
     def _should_create_constraint(self, compiler: Any, **kw: Any) -> bool:
         return bool(compiler.dialect.name == "oracle")
@@ -136,3 +143,7 @@
             _type_bound=True,
         )
         table.append_constraint(e)
+
+
+BigIntIdentity = BigInteger().with_variant(Integer, "sqlite")
+JsonB = _JSON().with_variant(PG_JSONB, "postgresql").with_variant(ORA_JSONB, "oracle")

{"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@@ -19,7 +19,7 @@\n     registry,\n )\n \n-from .types import GUID, JSON, BigIntIdentity\n+from .types import GUID, BigIntIdentity, JsonB\n \n if TYPE_CHECKING:\n     from sqlalchemy.sql import FromClause\n@@ -151,7 +151,7 @@\n             EmailStr: String,\n             AnyUrl: String,\n             AnyHttpUrl: String,\n-            dict: JSON,\n+            dict: JsonB,\n             datetime: DateTime,\n             date: Date,\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@@ -9,14 +9,14 @@\n from sqlalchemy.dialects.oracle import RAW as ORA_RAW\n from sqlalchemy.dialects.postgresql import JSONB as PG_JSONB\n from sqlalchemy.dialects.postgresql import UUID as PG_UUID\n-from sqlalchemy.types import BINARY, CHAR, BigInteger, Integer, SchemaType, TypeDecorator\n+from sqlalchemy.types import BINARY, CHAR, BigInteger, Integer, SchemaType, TypeDecorator, TypeEngine\n from sqlalchemy.types import JSON as _JSON\n \n+from litestar.serialization import decode_json, encode_json\n+\n if TYPE_CHECKING:\n     from sqlalchemy.engine import Dialect\n \n-BigIntIdentity = BigInteger().with_variant(Integer, \"sqlite\")\n-\n \n class GUID(TypeDecorator):\n     \"\"\"Platform-independent GUID type.\n@@ -81,16 +81,14 @@\n         return cast(\"uuid.UUID | None\", value)\n \n \n-class JSON(TypeDecorator, SchemaType):  # type: ignore\n-    \"\"\"Platform-independent JSON type.\n-\n-    Uses JSONB type for postgres, BLOB for Oracle, otherwise uses the generic JSON data type.\n+class ORA_JSONB(TypeDecorator, SchemaType):  # type: ignore  # noqa: N801\n+    \"\"\"Oracle Binary JSON type.\n \n-    JSON = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_BLOB, \"oracle\")\n+    JsonB = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_JSONB, \"oracle\")\n \n     \"\"\"\n \n-    impl = _JSON\n+    impl = ORA_BLOB\n     cache_ok = True\n \n     @property\n@@ -102,12 +100,21 @@\n         self.name = kwargs.pop(\"name\", None)\n         self.oracle_strict = kwargs.pop(\"oracle_strict\", True)\n \n-    def load_dialect_impl(self, dialect: Dialect) -> Any:\n-        if dialect.name == \"postgresql\":\n-            return dialect.type_descriptor(PG_JSONB())  # type: ignore\n-        if dialect.name == \"oracle\":\n-            return dialect.type_descriptor(ORA_BLOB())\n-        return dialect.type_descriptor(_JSON())\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@@ -136,3 +143,7 @@\n             _type_bound=True,\n         )\n         table.append_constraint(e)\n+\n+\n+BigIntIdentity = BigInteger().with_variant(Integer, \"sqlite\")\n+JsonB = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_JSONB, \"oracle\")\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, JSON, BigIntIdentity\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 = exclude.union(\"_sentinel\") if exclude else {\"_sentinel\"}\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: JSON,\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\nfrom sqlalchemy.types import JSON as _JSON\n\nif TYPE_CHECKING:\n    from sqlalchemy.engine import Dialect\n\nBigIntIdentity = BigInteger().with_variant(Integer, \"sqlite\")\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 JSON(TypeDecorator, SchemaType):  # type: ignore\n    \"\"\"Platform-independent JSON type.\n\n    Uses JSONB type for postgres, BLOB for Oracle, otherwise uses the generic JSON data type.\n\n    JSON = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_BLOB, \"oracle\")\n\n    \"\"\"\n\n    impl = _JSON\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 load_dialect_impl(self, dialect: Dialect) -> Any:\n        if dialect.name == \"postgresql\":\n            return dialect.type_descriptor(PG_JSONB())  # type: ignore\n        if dialect.name == \"oracle\":\n            return dialect.type_descriptor(ORA_BLOB())\n        return dialect.type_descriptor(_JSON())\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", "path": "litestar/contrib/sqlalchemy/types.py"}], "after_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 = exclude.union(\"_sentinel\") if exclude else {\"_sentinel\"}\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\")\nJsonB = _JSON().with_variant(PG_JSONB, \"postgresql\").with_variant(ORA_JSONB, \"oracle\")\n", "path": "litestar/contrib/sqlalchemy/types.py"}]}