1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-26 21:14:19 +00:00

Remove royalnet.engineer.bolts, which is not usable anymore after royalnet.alchemist was removed

This commit is contained in:
Steffo 2022-05-02 05:22:45 +02:00
parent 5ce523be1f
commit f71e4be9f3
Signed by: steffo
GPG key ID: 6965406171929D01

View file

@ -1,54 +0,0 @@
"""
This module contains **bolts**, utility decorators which can be used to enhance
:class:`~royalnet.engineer.conversation.Conversation`\\ s.
"""
from __future__ import annotations
import functools
import logging
import typing as t
import sqlalchemy.orm
import royalnet.lazy
log = logging.getLogger(__name__)
def use_database(session_class: t.Union[t.Type[sqlalchemy.orm.Session], royalnet.lazy.Lazy], *args, **kwargs):
"""
Decorator factory which allows a :class:`~royalnet.engineer.conversation.Conversation` to use a
:class:`sqlalchemy.orm.Session` created from the passed :class:`sqlalchemy.orm.sessionmaker` .
The session is automatically opened and closed, and will be available in the `_session` kwarg.
:param session_class: The :class:`sqlalchemy.orm.Session` class to use when creating the session.
It can also be provided wrapped in a :class:`royalnet.lazy.Lazy` object, from which it will
be evaluated.
:return: The decorator to use to decorate the function.
"""
if isinstance(session_class, royalnet.lazy.Lazy):
session_class = session_class.evaluate()
def decorator(f):
@functools.wraps(f)
async def decorated(**f_kwargs):
log.debug(f"Opening database session from {session_class!r}...")
with session_class(*args, **kwargs) as session:
log.debug(f"Opened database session {session!r} successfully!")
result = await f(**f_kwargs, _session=session)
log.debug(f"Closing database session {session!r}...")
log.debug(f"Closed database session from {session_class!r} successfully!")
# Shouldn't be necessary, conversations return None anyways
return result
return decorated
return decorator
__all__ = (
"use_database",
)