UnitOfWork

Bases: Protocol

Unit of Work protocol defining the interface for a unit of work implementation.

Source code in src/alpha/interfaces/unit_of_work.py
@runtime_checkable
class UnitOfWork(Protocol):
    """Unit of Work protocol defining the interface for a unit of work 
    implementation.
    """

    def __enter__(self: UOW) -> UOW:
        """Enter the runtime context related to this object.

        Returns
        -------
        UOW
            The unit of work instance.
        """
        ...

    def __exit__(self, *args: Any) -> None:
        """Exit the runtime context related to this object.
        """
        ...

    def commit(self) -> None:
        """Commit the current transaction."""
        ...

    def flush(self) -> None:
        """Flush the current session."""
        ...

    def rollback(self) -> None:
        """Rollback the current transaction."""
        ...

    def refresh(self, obj: object) -> None:
        """Refresh the state of the given object from the database.

        Parameters
        ----------
        obj
            The object to refresh.
        """
        ...

    @property
    def session(self) -> Session:
        """Get the current session.

        Returns
        -------
        Session
            The current SQLAlchemy session.
        """
        ...

session property

session

Get the current session.

Returns:
  • Session

    The current SQLAlchemy session.

Methods:

__enter__

__enter__()

Enter the runtime context related to this object.

Returns:
  • UOW

    The unit of work instance.

Source code in src/alpha/interfaces/unit_of_work.py
def __enter__(self: UOW) -> UOW:
    """Enter the runtime context related to this object.

    Returns
    -------
    UOW
        The unit of work instance.
    """
    ...

__exit__

__exit__(*args)

Exit the runtime context related to this object.

Source code in src/alpha/interfaces/unit_of_work.py
def __exit__(self, *args: Any) -> None:
    """Exit the runtime context related to this object.
    """
    ...

commit

commit()

Commit the current transaction.

Source code in src/alpha/interfaces/unit_of_work.py
def commit(self) -> None:
    """Commit the current transaction."""
    ...

flush

flush()

Flush the current session.

Source code in src/alpha/interfaces/unit_of_work.py
def flush(self) -> None:
    """Flush the current session."""
    ...

rollback

rollback()

Rollback the current transaction.

Source code in src/alpha/interfaces/unit_of_work.py
def rollback(self) -> None:
    """Rollback the current transaction."""
    ...

refresh

refresh(obj)

Refresh the state of the given object from the database.

Parameters:
  • obj (object) –

    The object to refresh.

Source code in src/alpha/interfaces/unit_of_work.py
def refresh(self, obj: object) -> None:
    """Refresh the state of the given object from the database.

    Parameters
    ----------
    obj
        The object to refresh.
    """
    ...