"""
    sa模块
"""
from typing import Any, Type, Union

from sqlalchemy.exc import ResourceClosedError

from smorest_sfs.extensions import db

from .query import SAQuery
from .statement import SAStatement


def _execute_sql(sa_sql: Any) -> Any:
    cursor = db.session.execute(sa_sql)
    try:
        return cursor.fetchall()
    except ResourceClosedError:
        pass


def _execute_sa(sql: SAStatement) -> Any:
    sa_sql = sql.get_sa_sql()
    return _execute_sql(sa_sql)


def _execute_query(query: SAQuery) -> Any:
    return query.get_record()


def execute(
    sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any, **kwargs: Any
) -> Any:
    sql = sql_cls(*args, **kwargs)
    if isinstance(sql, SAQuery):
        ret = _execute_query(sql)
    else:
        ret = _execute_sa(sql)

    return ret


def debug_sql(
    sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any, **kwargs: Any
) -> None:
    sql = sql_cls(*args, **kwargs)
    sql.debug_sql()


def render_limit_results(
    sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any, **kwargs: Any
) -> None:
    sql = sql_cls(*args, **kwargs)
    sql.render_results()

3abfb1a0a784f3d4d3fa80b5b74cb4efe422cfdd
<mask token>


def _execute_sql(sa_sql: Any) ->Any:
    cursor = db.session.execute(sa_sql)
    try:
        return cursor.fetchall()
    except ResourceClosedError:
        pass


<mask token>


def _execute_query(query: SAQuery) ->Any:
    return query.get_record()


def execute(sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any,
    **kwargs: Any) ->Any:
    sql = sql_cls(*args, **kwargs)
    if isinstance(sql, SAQuery):
        ret = _execute_query(sql)
    else:
        ret = _execute_sa(sql)
    return ret


<mask token>

<mask token>


def _execute_sql(sa_sql: Any) ->Any:
    cursor = db.session.execute(sa_sql)
    try:
        return cursor.fetchall()
    except ResourceClosedError:
        pass


def _execute_sa(sql: SAStatement) ->Any:
    sa_sql = sql.get_sa_sql()
    return _execute_sql(sa_sql)


def _execute_query(query: SAQuery) ->Any:
    return query.get_record()


def execute(sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any,
    **kwargs: Any) ->Any:
    sql = sql_cls(*args, **kwargs)
    if isinstance(sql, SAQuery):
        ret = _execute_query(sql)
    else:
        ret = _execute_sa(sql)
    return ret


<mask token>

<mask token>


def _execute_sql(sa_sql: Any) ->Any:
    cursor = db.session.execute(sa_sql)
    try:
        return cursor.fetchall()
    except ResourceClosedError:
        pass


def _execute_sa(sql: SAStatement) ->Any:
    sa_sql = sql.get_sa_sql()
    return _execute_sql(sa_sql)


def _execute_query(query: SAQuery) ->Any:
    return query.get_record()


def execute(sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any,
    **kwargs: Any) ->Any:
    sql = sql_cls(*args, **kwargs)
    if isinstance(sql, SAQuery):
        ret = _execute_query(sql)
    else:
        ret = _execute_sa(sql)
    return ret


def debug_sql(sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any,
    **kwargs: Any) ->None:
    sql = sql_cls(*args, **kwargs)
    sql.debug_sql()


<mask token>

<mask token>


def _execute_sql(sa_sql: Any) ->Any:
    cursor = db.session.execute(sa_sql)
    try:
        return cursor.fetchall()
    except ResourceClosedError:
        pass


def _execute_sa(sql: SAStatement) ->Any:
    sa_sql = sql.get_sa_sql()
    return _execute_sql(sa_sql)


def _execute_query(query: SAQuery) ->Any:
    return query.get_record()


def execute(sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any,
    **kwargs: Any) ->Any:
    sql = sql_cls(*args, **kwargs)
    if isinstance(sql, SAQuery):
        ret = _execute_query(sql)
    else:
        ret = _execute_sa(sql)
    return ret


def debug_sql(sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any,
    **kwargs: Any) ->None:
    sql = sql_cls(*args, **kwargs)
    sql.debug_sql()


def render_limit_results(sql_cls: Union[Type[SAStatement], Type[SAQuery]],
    *args: Any, **kwargs: Any) ->None:
    sql = sql_cls(*args, **kwargs)
    sql.render_results()

"""
    sa模块
"""
from typing import Any, Type, Union

from sqlalchemy.exc import ResourceClosedError

from smorest_sfs.extensions import db

from .query import SAQuery
from .statement import SAStatement


def _execute_sql(sa_sql: Any) -> Any:
    cursor = db.session.execute(sa_sql)
    try:
        return cursor.fetchall()
    except ResourceClosedError:
        pass


def _execute_sa(sql: SAStatement) -> Any:
    sa_sql = sql.get_sa_sql()
    return _execute_sql(sa_sql)


def _execute_query(query: SAQuery) -> Any:
    return query.get_record()


def execute(
    sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any, **kwargs: Any
) -> Any:
    sql = sql_cls(*args, **kwargs)
    if isinstance(sql, SAQuery):
        ret = _execute_query(sql)
    else:
        ret = _execute_sa(sql)

    return ret


def debug_sql(
    sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any, **kwargs: Any
) -> None:
    sql = sql_cls(*args, **kwargs)
    sql.debug_sql()


def render_limit_results(
    sql_cls: Union[Type[SAStatement], Type[SAQuery]], *args: Any, **kwargs: Any
) -> None:
    sql = sql_cls(*args, **kwargs)
    sql.render_results()
