File: test_sqlalchemy.py

package info (click to toggle)
litestar 2.21.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 12,568 kB
  • sloc: python: 70,588; makefile: 254; javascript: 104; sh: 60
file content (375 lines) | stat: -rw-r--r-- 18,327 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
# ruff: noqa: F401
# pyright: reportUnusedImport=false
from __future__ import annotations

import importlib
import sys
from pathlib import Path

import pytest
from advanced_alchemy import exceptions as advanced_alchemy_exceptions
from advanced_alchemy import repository as advanced_alchemy_repo
from advanced_alchemy import types as advanced_alchemy_types
from advanced_alchemy.repository import typing as advanced_alchemy_typing
from sqlalchemy import Engine, create_engine
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine


def purge_module(module_names: list[str], path: str | Path) -> None:
    for name in module_names:
        if name in sys.modules:
            del sys.modules[name]
    Path(importlib.util.cache_from_source(path)).unlink(missing_ok=True)  # type: ignore[arg-type]


def test_create_engine_with_engine_instance() -> None:
    from litestar.contrib.sqlalchemy.plugins.init.config.sync import SQLAlchemySyncConfig

    engine = create_engine("sqlite:///:memory:")
    config = SQLAlchemySyncConfig(engine_instance=engine)
    with pytest.deprecated_call():
        assert engine is config.create_engine()  # type: ignore[attr-defined]


def test_create_engine_with_connection_string() -> None:
    from litestar.contrib.sqlalchemy.plugins.init.config.sync import SQLAlchemySyncConfig

    config = SQLAlchemySyncConfig(connection_string="sqlite:///:memory:")
    with pytest.deprecated_call():
        engine = config.create_engine()  # type: ignore[attr-defined]
    assert isinstance(engine, Engine)


def test_async_create_engine_with_engine_instance() -> None:
    from litestar.contrib.sqlalchemy.plugins.init.config.asyncio import SQLAlchemyAsyncConfig

    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    config = SQLAlchemyAsyncConfig(engine_instance=engine)
    with pytest.deprecated_call():
        assert engine is config.create_engine()  # type: ignore[attr-defined]


def test_async_create_engine_with_connection_string() -> None:
    from litestar.contrib.sqlalchemy.plugins.init.config.asyncio import SQLAlchemyAsyncConfig

    config = SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///:memory:")
    with pytest.deprecated_call():
        engine = config.create_engine()  # type: ignore[attr-defined]
    assert isinstance(engine, AsyncEngine)


def test_repository_re_exports() -> None:
    from litestar.contrib.sqlalchemy import types
    from litestar.contrib.sqlalchemy.repository import (
        SQLAlchemyAsyncRepository,
        SQLAlchemySyncRepository,
        wrap_sqlalchemy_exception,
    )
    from litestar.contrib.sqlalchemy.repository import types as repository_types

    assert wrap_sqlalchemy_exception is advanced_alchemy_exceptions.wrap_sqlalchemy_exception
    assert SQLAlchemySyncRepository is advanced_alchemy_repo.SQLAlchemySyncRepository
    assert SQLAlchemyAsyncRepository is advanced_alchemy_repo.SQLAlchemyAsyncRepository
    assert repository_types.ModelT is advanced_alchemy_typing.ModelT  # pyright: ignore[reportGeneralTypeIssues]
    assert repository_types.RowT is advanced_alchemy_typing.RowT  # pyright: ignore[reportGeneralTypeIssues]
    assert repository_types.SQLAlchemyAsyncRepositoryT is advanced_alchemy_typing.SQLAlchemyAsyncRepositoryT  # pyright: ignore[reportGeneralTypeIssues]
    assert repository_types.SQLAlchemySyncRepositoryT is advanced_alchemy_typing.SQLAlchemySyncRepositoryT  # pyright: ignore[reportGeneralTypeIssues]

    assert types.GUID is advanced_alchemy_types.GUID
    assert types.ORA_JSONB is advanced_alchemy_types.ORA_JSONB
    assert types.BigIntIdentity is advanced_alchemy_types.BigIntIdentity
    assert types.DateTimeUTC is advanced_alchemy_types.DateTimeUTC
    assert types.JsonB is advanced_alchemy_types.JsonB


def test_deprecated_sqlalchemy_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy"], __file__)
    with pytest.warns(
        DeprecationWarning, match="importing SQLAlchemyAsyncRepository from 'litestar.contrib.sqlalchemy' is deprecated"
    ):
        from litestar.contrib.sqlalchemy import SQLAlchemyAsyncRepository
    purge_module(["litestar.contrib.sqlalchemy"], __file__)
    with pytest.warns(
        DeprecationWarning, match="importing SQLAlchemySyncRepository from 'litestar.contrib.sqlalchemy' is deprecated"
    ):
        from litestar.contrib.sqlalchemy import SQLAlchemySyncRepository
    purge_module(["litestar.contrib.sqlalchemy"], __file__)
    with pytest.warns(DeprecationWarning, match="importing ModelT from 'litestar.contrib.sqlalchemy' is deprecated"):
        from litestar.contrib.sqlalchemy import ModelT
    purge_module(["litestar.contrib.sqlalchemy"], __file__)
    with pytest.warns(
        DeprecationWarning, match="importing wrap_sqlalchemy_exception from 'litestar.contrib.sqlalchemy' is deprecated"
    ):
        from litestar.contrib.sqlalchemy import wrap_sqlalchemy_exception


def test_deprecated_sqlalchemy_plugins_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing AsyncSessionConfig from 'litestar.contrib.sqlalchemy.plugins' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins import AsyncSessionConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins"], __file__)
    with pytest.warns(
        DeprecationWarning, match="importing EngineConfig from 'litestar.contrib.sqlalchemy.plugins' is deprecated"
    ):
        from litestar.contrib.sqlalchemy.plugins import EngineConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing GenericSQLAlchemyConfig from 'litestar.contrib.sqlalchemy.plugins' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins import GenericSQLAlchemyConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyAsyncConfig from 'litestar.contrib.sqlalchemy.plugins' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins import SQLAlchemyAsyncConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyInitPlugin from 'litestar.contrib.sqlalchemy.plugins' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins import SQLAlchemyInitPlugin


def test_deprecated_sqlalchemy_plugins_init_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing AsyncSessionConfig from 'litestar.contrib.sqlalchemy.plugins.init' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init import AsyncSessionConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init"], __file__)
    with pytest.warns(
        DeprecationWarning, match="importing EngineConfig from 'litestar.contrib.sqlalchemy.plugins.init' is deprecated"
    ):
        from litestar.contrib.sqlalchemy.plugins.init import EngineConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing GenericSQLAlchemyConfig from 'litestar.contrib.sqlalchemy.plugins.init' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init import GenericSQLAlchemyConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyAsyncConfig from 'litestar.contrib.sqlalchemy.plugins.init' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init import SQLAlchemyAsyncConfig


def test_deprecated_sqlalchemy_plugins_init_config_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing AsyncSessionConfig from 'litestar.contrib.sqlalchemy.plugins.init.config' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config import AsyncSessionConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing EngineConfig from 'litestar.contrib.sqlalchemy.plugins.init.config' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config import EngineConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing GenericSQLAlchemyConfig from 'litestar.contrib.sqlalchemy.plugins.init.config' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config import GenericSQLAlchemyConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyAsyncConfig from 'litestar.contrib.sqlalchemy.plugins.init.config' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config import SQLAlchemyAsyncConfig


def test_deprecated_sqlalchemy_plugins_init_config_common_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.common"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SESSION_SCOPE_KEY from 'litestar.contrib.sqlalchemy.plugins.init.config.common' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.common import SESSION_SCOPE_KEY
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.common"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SESSION_TERMINUS_ASGI_EVENTS from 'litestar.contrib.sqlalchemy.plugins.init.config.common' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.common import SESSION_TERMINUS_ASGI_EVENTS
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.common"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing GenericSQLAlchemyConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.common' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.common import GenericSQLAlchemyConfig


def test_deprecated_sqlalchemy_plugins_init_config_sync_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.sync"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemySyncConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.sync' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.sync import SQLAlchemySyncConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.sync"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing AlembicSyncConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.sync' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.sync import AlembicSyncConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.sync"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SyncSessionConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.sync' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.sync import SyncSessionConfig


def test_deprecated_sqlalchemy_plugins_init_config_asyncio_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.asyncio"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyAsyncConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.asyncio' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.asyncio import SQLAlchemyAsyncConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.asyncio"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing AlembicAsyncConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.asyncio' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.asyncio import AlembicAsyncConfig
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.asyncio"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing AsyncSessionConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.asyncio' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.asyncio import AsyncSessionConfig


def test_deprecated_sqlalchemy_plugins_init_config_engine_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.engine"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing EngineConfig from 'litestar.contrib.sqlalchemy.plugins.init.config.engine' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.engine import EngineConfig


def test_deprecated_sqlalchemy_dto_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.dto"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyDTOConfig from 'litestar.contrib.sqlalchemy.dto' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTOConfig


def test_deprecated_sqlalchemy_plugins_init_plugin_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.plugin"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyInitPlugin from 'litestar.contrib.sqlalchemy.plugins.init' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.plugin import SQLAlchemyInitPlugin


def test_deprecated_sqlalchemy_plugins_serialization_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.serialization"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemySerializationPlugin from 'litestar.contrib.sqlalchemy.plugins.serialization' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.serialization import SQLAlchemySerializationPlugin


def test_deprecated_sqlalchemy_repository_async_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.repository._async"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemyAsyncRepository from 'litestar.contrib.sqlalchemy.repository._async' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.repository._async import SQLAlchemyAsyncRepository


def test_deprecated_sqlalchemy_repository_sync_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.repository._sync"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing SQLAlchemySyncRepository from 'litestar.contrib.sqlalchemy.repository._sync' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.repository._sync import SQLAlchemySyncRepository


def test_deprecated_sqlalchemy_base_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.base"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="from 'litestar.contrib.sqlalchemy.base' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.base import (
            AuditColumns,
            BigIntAuditBase,
            BigIntBase,
            BigIntPrimaryKey,
            CommonTableAttributes,
            ModelProtocol,
            UUIDAuditBase,
            UUIDBase,
            UUIDPrimaryKey,
            create_registry,
            orm_registry,
            touch_updated_timestamp,
        )


def test_deprecated_sqlalchemy_plugins_init_config_asyncio_handlers() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.asyncio"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing default_before_send_handler from 'litestar.contrib.sqlalchemy.plugins.init.config.asyncio' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.asyncio import default_before_send_handler

    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.asyncio"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing autocommit_before_send_handler from 'litestar.contrib.sqlalchemy.plugins.init.config.asyncio' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.asyncio import autocommit_before_send_handler


def test_deprecated_sqlalchemy_plugins_init_config_sync_handlers() -> None:
    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.sync"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing default_before_send_handler from 'litestar.contrib.sqlalchemy.plugins.init.config.sync' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.sync import default_before_send_handler

    purge_module(["litestar.contrib.sqlalchemy.plugins.init.config.sync"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing autocommit_before_send_handler from 'litestar.contrib.sqlalchemy.plugins.init.config.sync' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.plugins.init.config.sync import autocommit_before_send_handler


def test_deprecated_sqlalchemy_repository_util_imports() -> None:
    purge_module(["litestar.contrib.sqlalchemy.repository._util"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing wrap_sqlalchemy_exception from 'litestar.contrib.sqlalchemy.repository._util' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.repository._util import wrap_sqlalchemy_exception

    purge_module(["litestar.contrib.sqlalchemy.repository._util"], __file__)
    with pytest.warns(
        DeprecationWarning,
        match="importing get_instrumented_attr from 'litestar.contrib.sqlalchemy.repository._util' is deprecated",
    ):
        from litestar.contrib.sqlalchemy.repository._util import get_instrumented_attr