aiocypher

29class Driver (object, metaclass=AIOCypherABCMeta):
30    """This class encapsulates a driver object for a cypher database. Concrete subclasses specialise in
31    different databases.
32    """
33    def __init__(self,
34                 config: Config):
35        self._address = config.address
36        self._auth: Tuple[str, str] = config.auth
37
38    async def __aenter__(self) -> 'Driver':
39        return self
40
41    async def __aexit__(self, e_t, e_v, e_tb) -> Optional[bool]:
42        await self.close()
43        return False
44
45    @abstractmethod
46    def __await__(self):
47        ...
48
49    @abstractmethod
50    async def close(self) -> None:
51        """This coroutine should be awaited when the driver is no longer needed.
52        """
53        ...
54
55    @abstractmethod
56    def session(self, **kwargs) -> AsyncContextManager[Session]:
57        """This method is used to create a Session object, which can be used as an
58        asynchronous context manager.
59
60        All interactions should be done via a Session object. Accessing the driver
61        without one is not supported.
62        """
63        ...
64
65    @abstractmethod
66    def database_type(self) -> str:
67        ...

25class Session (object, metaclass=ABCMeta):
26    """This object encapsulates a Session object. It should not be initialised directly, but
27    created via a Driver object.
28
29    It should always be used as an Async Context Manager, and using its methods outside of its
30    context may lead to undefined behaviour.
31
32    A Session object is the only supported way to interact with the cypher database.
33    """
34    async def __aenter__(self) -> 'Session':
35        return self
36
37    async def __aexit__(self, *args) -> bool:
38        return False
39
40    @abstractmethod
41    def begin_transaction(self) -> AsyncContextManager[Transaction]:
42        """This is the main method for interacting with the Session. It
43        creates a new Transaction which is then represented by an Async Context Manager
44        """
45        ...

24class Transaction (object, metaclass=ABCMeta):
25    """A class which encapsulates a Transaction. Should never be created directly,
26    but always via a Session object.
27
28    Should always be used as an Async Context Manager. Using any of its methods outside of
29    the context may lead to undefined behaviour.
30    """
31    async def __aenter__(self) -> "Transaction":
32        return self
33
34    async def __aexit__(self, *args):
35        return False
36
37    @abstractmethod
38    def run(self, *args, **kwargs) -> Result:
39        """This is the main routine for this class.
40
41        The actual query will not be executed until such a time as the return value (or some derivative thereof)
42        is awaited.
43
44        :returns: an Result object
45        """
46        ...

30class Result (Generic[R], Awaitable[R], metaclass=ABCMeta):
31    """A conceptual wrapper for a query which will return a result object of some kind.
32
33    To execute the query and return the underlying object await this object.
34
35    It also has other methods which return awaitable objects.
36    """
37
38    @abstractmethod
39    def single(self) -> Record:
40        """Returns a Record object
41
42        Useful if the wrapped query is expected to return a single result.
43        """
44        ...
45
46    @abstractmethod
47    async def data(self) -> List[Dict[str, Any]]:
48        """Deserialises the data in this result as a list of dictionaries and returns it.
49        """
50        ...
51
52    @abstractmethod
53    def graph(self) -> Graph:
54        """Returns an Graph object which
55
56        Useful if the wrapped query is expected to return multiple results.
57        """
58        ...

26class Relationship (Mapping[str, Union[str, int]], metaclass=ABCMeta):
27    type: str
28    start_node: Node
29    end_node: Node

24class Node (Mapping[str, Union[str, int, List[Union[str, int]]]], metaclass=ABCMeta):
25    labels: Set[str]

43class EmptyResult(Result[None]):
44    def single(self) -> Record:
45        """Returns a Record object
46
47        Useful if the wrapped query is expected to return a single result.
48        """
49        raise RuntimeError("This result is empty")
50
51    async def data(self) -> List[Dict[str, Any]]:
52        """Deserialises the data in this result as a list of dictionaries and returns it.
53        """
54        return []
55
56    def graph(self) -> EmptyGraph:
57        """Returns an Graph object which
58
59        Useful if the wrapped query is expected to return multiple results.
60        """
61        return EmptyGraph()
62
63    def __await__(self):
64        async def __inner():
65            return None
66        return __inner().__await__()

22class QueryFailed (Exception):
23    def __init__(self, query: str, params: Dict[str, Union[str, int]]):
24        super().__init__(f"Query failed: << {query} >> with parameters {params!r}")

22class Config(NamedTuple):
23    address: str
24    username: str
25    password: str
26
27    @property
28    def auth(self) -> Tuple[str, str]:
29        return (self.username, self.password)