fauna.client.utils

 1import os
 2import threading
 3from typing import Generic, Callable, TypeVar, Optional
 4
 5from fauna.client.endpoints import Endpoints
 6from fauna.client.headers import Header
 7
 8
 9def _fancy_bool_from_str(val: str) -> bool:
10  return val.lower() in ["1", "true", "yes", "y"]
11
12
13class LastTxnTs(object):
14  """Wraps tracking the last transaction time supplied from the database."""
15
16  def __init__(
17      self,
18      time: Optional[int] = None,
19  ):
20    self._lock: threading.Lock = threading.Lock()
21    self._time: Optional[int] = time
22
23  @property
24  def time(self):
25    """Produces the last transaction time, or, None if not yet updated."""
26    with self._lock:
27      return self._time
28
29  @property
30  def request_header(self):
31    """Produces a dictionary with a non-zero `X-Last-Seen-Txn` header; or,
32        if one has not yet been set, the empty header dictionary."""
33    t = self._time
34    if t is None:
35      return {}
36    return {Header.LastTxnTs: str(t)}
37
38  def update_txn_time(self, new_txn_time: int):
39    """Updates the internal transaction time.
40        In order to maintain a monotonically-increasing value, `newTxnTime`
41        is discarded if it is behind the current timestamp."""
42    with self._lock:
43      self._time = max(self._time or 0, new_txn_time)
44
45
46T = TypeVar('T')
47
48
49class _SettingFromEnviron(Generic[T]):
50
51  def __init__(
52      self,
53      var_name: str,
54      default_value: str,
55      adapt_from_str: Callable[[str], T],
56  ):
57    self.__var_name = var_name
58    self.__default_value = default_value
59    self.__adapt_from_str = adapt_from_str
60
61  def __call__(self) -> T:
62    return self.__adapt_from_str(
63        os.environ.get(
64            self.__var_name,
65            default=self.__default_value,
66        ))
67
68
69class _Environment:
70  EnvFaunaEndpoint = _SettingFromEnviron(
71      "FAUNA_ENDPOINT",
72      Endpoints.Default,
73      str,
74  )
75  """environment variable for Fauna Client HTTP endpoint"""
76
77  EnvFaunaSecret = _SettingFromEnviron(
78      "FAUNA_SECRET",
79      "",
80      str,
81  )
82  """environment variable for Fauna Client authentication"""