ellar-sql/ellar_sql/query/utils.py at 5ac87a6404016a08e3317ef85a153b5c54c9aaa8 · python-ellar/ellar-sql · GitHub

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
import typing as t

import ellar.common as ecm
import sqlalchemy as sa
import sqlalchemy.exc as sa_exc
from ellar.core import current_injector

from ellar_sql.services import EllarSQLService

_O = t.TypeVar("_O", bound=object)


async def get_or_404(
    entity: t.Type[_O],
    ident: t.Any,
    *,
    error_message: t.Optional[str] = None,
    **kwargs: t.Any,
) -> _O:
    """ """
    db_service = current_injector.get(EllarSQLService)
    session = db_service.session_factory_maker()()

    value = session.get(entity, ident, **kwargs)

    if isinstance(value, t.Coroutine):
        value = await value

    if value is None:
        raise ecm.NotFound(detail=error_message)

    return t.cast(_O, value)


async def get_or_none(
    entity: t.Type[_O],
    ident: t.Any,
    **kwargs: t.Any,
) -> t.Optional[_O]:
    """ """
    db_service = current_injector.get(EllarSQLService)
    session = db_service.session_factory_maker()()

    value = session.get(entity, ident, **kwargs)

    if isinstance(value, t.Coroutine):
        value = await value

    if value is None:
        return None

    return t.cast(_O, value)


async def first_or_404(
    statement: sa.sql.Select[t.Any], *, error_message: t.Optional[str] = None
) -> t.Any:
    """ """
    db_service = current_injector.get(EllarSQLService)
    session = db_service.session_factory()

    result = session.execute(statement)
    if isinstance(result, t.Coroutine):
        result = await result

    value = result.scalar()

    if value is None:
        raise ecm.NotFound(detail=error_message)

    return value


async def first_or_none(statement: sa.sql.Select[t.Any]) -> t.Any:
    """ """
    db_service = current_injector.get(EllarSQLService)
    session = db_service.session_factory()

    result = session.execute(statement)
    if isinstance(result, t.Coroutine):
        result = await result

    value = result.scalar()

    if value is None:
        return None

    return value


async def one_or_404(
    statement: sa.sql.Select[t.Any], *, error_message: t.Optional[str] = None
) -> t.Any:
    """ """
    db_service = current_injector.get(EllarSQLService)
    session = db_service.session_factory()

    try:
        result = session.execute(statement)

        if isinstance(result, t.Coroutine):
            result = await result

        return result.scalar_one()
    except (sa_exc.NoResultFound, sa_exc.MultipleResultsFound) as ex:
        raise ecm.NotFound(detail=error_message) from ex