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
|
import pytest
from sqlalchemy.exc import (
IntegrityError as SQLAlchemyIntegrityError,
)
from sqlalchemy.exc import (
InvalidRequestError as SQLAlchemyInvalidRequestError,
)
from sqlalchemy.exc import (
MultipleResultsFound,
SQLAlchemyError,
StatementError,
)
from advanced_alchemy.exceptions import (
DuplicateKeyError,
IntegrityError,
InvalidRequestError,
MultipleResultsFoundError,
NotFoundError,
RepositoryError,
wrap_sqlalchemy_exception,
)
def test_wrap_sqlalchemy_exception_multiple_results_found() -> None:
with pytest.raises(MultipleResultsFoundError), wrap_sqlalchemy_exception():
raise MultipleResultsFound()
@pytest.mark.parametrize("dialect_name", ["postgresql", "sqlite", "mysql"])
def test_wrap_sqlalchemy_exception_integrity_error_duplicate_key(dialect_name: str) -> None:
error_message = {
"postgresql": 'duplicate key value violates unique constraint "uq_%(table_name)s_%(column_0_name)s"',
"sqlite": "UNIQUE constraint failed: %(table_name)s.%(column_0_name)s",
"mysql": "1062 (23000): Duplicate entry '%(value)s' for key '%(table_name)s.%(column_0_name)s'",
}
with (
pytest.raises(DuplicateKeyError),
wrap_sqlalchemy_exception(
dialect_name=dialect_name,
error_messages={"duplicate_key": error_message[dialect_name]},
),
):
if dialect_name == "postgresql":
exception = SQLAlchemyIntegrityError(
"INSERT INTO table (id) VALUES (1)",
{"table_name": "table", "column_0_name": "id"},
Exception(
'duplicate key value violates unique constraint "uq_table_id"\nDETAIL: Key (id)=(1) already exists.',
),
)
elif dialect_name == "sqlite":
exception = SQLAlchemyIntegrityError(
"INSERT INTO table (id) VALUES (1)",
{"table_name": "table", "column_0_name": "id"},
Exception("UNIQUE constraint failed: table.id"),
)
else:
exception = SQLAlchemyIntegrityError(
"INSERT INTO table (id) VALUES (1)",
{"table_name": "table", "column_0_name": "id", "value": "1"},
Exception("1062 (23000): Duplicate entry '1' for key 'table.id'"),
)
raise exception
def test_wrap_sqlalchemy_exception_integrity_error_other() -> None:
with pytest.raises(IntegrityError), wrap_sqlalchemy_exception():
raise SQLAlchemyIntegrityError("original", {}, Exception("original"))
def test_wrap_sqlalchemy_exception_invalid_request_error() -> None:
with pytest.raises(InvalidRequestError), wrap_sqlalchemy_exception():
raise SQLAlchemyInvalidRequestError("original", {}, Exception("original"))
def test_wrap_sqlalchemy_exception_statement_error() -> None:
with pytest.raises(IntegrityError), wrap_sqlalchemy_exception():
raise StatementError("original", None, {}, Exception("original")) # pyright: ignore[reportArgumentType]
def test_wrap_sqlalchemy_exception_sqlalchemy_error() -> None:
with pytest.raises(RepositoryError), wrap_sqlalchemy_exception():
raise SQLAlchemyError("original")
def test_wrap_sqlalchemy_exception_attribute_error() -> None:
with pytest.raises(RepositoryError), wrap_sqlalchemy_exception():
raise AttributeError("original")
def test_wrap_sqlalchemy_exception_not_found_error() -> None:
with pytest.raises(NotFoundError, match="No rows matched the specified data"), wrap_sqlalchemy_exception():
raise NotFoundError("No item found when one was expected")
def test_wrap_sqlalchemy_exception_no_wrap() -> None:
with pytest.raises(SQLAlchemyError), wrap_sqlalchemy_exception(wrap_exceptions=False):
raise SQLAlchemyError("original")
with pytest.raises(SQLAlchemyIntegrityError), wrap_sqlalchemy_exception(wrap_exceptions=False):
raise SQLAlchemyIntegrityError(statement="select 1", params=None, orig=BaseException())
with pytest.raises(MultipleResultsFound), wrap_sqlalchemy_exception(wrap_exceptions=False):
raise MultipleResultsFound()
with pytest.raises(SQLAlchemyInvalidRequestError), wrap_sqlalchemy_exception(wrap_exceptions=False):
raise SQLAlchemyInvalidRequestError()
with pytest.raises(AttributeError), wrap_sqlalchemy_exception(wrap_exceptions=False):
raise AttributeError()
with (
pytest.raises(NotFoundError, match="No item found when one was expected"),
wrap_sqlalchemy_exception(wrap_exceptions=False),
):
raise NotFoundError("No item found when one was expected")
def test_custom_not_found_error_message() -> None:
with (
pytest.raises(NotFoundError, match="Custom Error"),
wrap_sqlalchemy_exception(error_messages={"not_found": "Custom Error"}),
):
raise NotFoundError("original")
def test_wrap_sqlalchemy_exception_custom_error_message() -> None:
def custom_message(exc: Exception) -> str:
return f"Custom: {exc}"
with (
pytest.raises(RepositoryError) as excinfo,
wrap_sqlalchemy_exception(
error_messages={"other": custom_message},
),
):
raise SQLAlchemyError("original")
assert str(excinfo.value) == "Custom: original"
def test_wrap_sqlalchemy_exception_no_error_messages() -> None:
with pytest.raises(RepositoryError) as excinfo, wrap_sqlalchemy_exception():
raise SQLAlchemyError("original")
assert str(excinfo.value) == "An exception occurred: original"
def test_wrap_sqlalchemy_exception_no_match() -> None:
with (
pytest.raises(IntegrityError) as excinfo,
wrap_sqlalchemy_exception(
dialect_name="postgresql",
error_messages={"integrity": "Integrity error"},
),
):
raise SQLAlchemyIntegrityError("original", {}, Exception("original"))
assert str(excinfo.value) == "Integrity error"
|
