fauna.client.headers

  1import os
  2import platform
  3import sys
  4from dataclasses import dataclass
  5from typing import Callable
  6
  7from fauna import __version__
  8
  9
 10class Header:
 11  LastTxnTs = "X-Last-Txn-Ts"
 12  Linearized = "X-Linearized"
 13  MaxContentionRetries = "X-Max-Contention-Retries"
 14  QueryTimeoutMs = "X-Query-Timeout-Ms"
 15  Typecheck = "X-Typecheck"
 16  Tags = "X-Query-Tags"
 17  Traceparent = "Traceparent"
 18
 19
 20class _Header:
 21  AcceptEncoding = "Accept-Encoding"
 22  Authorization = "Authorization"
 23  ContentType = "Content-Type"
 24  Driver = "X-Driver"
 25  DriverEnv = "X-Driver-Env"
 26  Format = "X-Format"
 27
 28
 29class _Auth:
 30  """Creates an auth helper object"""
 31
 32  def bearer(self):
 33    return "Bearer {}".format(self.secret)
 34
 35  def __init__(self, secret):
 36    self.secret = secret
 37
 38  def __eq__(self, other):
 39    return self.secret == getattr(other, 'secret', None)
 40
 41  def __ne__(self, other):
 42    return not self == other
 43
 44
 45class _DriverEnvironment:
 46
 47  def __init__(self):
 48    self.pythonVersion = "{0}.{1}.{2}-{3}".format(*sys.version_info)
 49    self.driverVersion = __version__
 50    self.env = self._get_runtime_env()
 51    self.os = "{0}-{1}".format(platform.system(), platform.release())
 52
 53  @staticmethod
 54  def _get_runtime_env():
 55
 56    @dataclass
 57    class EnvChecker:
 58      name: str
 59      check: Callable[[], bool]
 60
 61    env: list[EnvChecker] = [
 62        EnvChecker(
 63            name="Netlify",
 64            check=lambda: "NETLIFY_IMAGES_CDN_DOMAIN" in os.environ,
 65        ),
 66        EnvChecker(
 67            name="Vercel",
 68            check=lambda: "VERCEL" in os.environ,
 69        ),
 70        EnvChecker(
 71            name="Heroku",
 72            check=lambda: "PATH" in \
 73                os.environ and ".heroku" in os.environ["PATH"],
 74        ),
 75        EnvChecker(
 76            name="AWS Lambda",
 77            check=lambda: "AWS_LAMBDA_FUNCTION_VERSION" in os.environ,
 78        ),
 79        EnvChecker(
 80            name="GCP Cloud Functions",
 81            check=lambda: "_" in \
 82                os.environ and "google" in os.environ["_"],
 83        ),
 84        EnvChecker(
 85            name="GCP Compute Instances",
 86            check=lambda: "GOOGLE_CLOUD_PROJECT" in os.environ,
 87        ),
 88        EnvChecker(
 89            name="Azure Cloud Functions",
 90            check=lambda: "WEBSITE_FUNCTIONS_AZUREMONITOR_CATEGORIES" in \
 91                os.environ,
 92        ),
 93        EnvChecker(
 94            name="Azure Compute",
 95            check=lambda: "ORYX_ENV_TYPE" in os.environ and \
 96                "WEBSITE_INSTANCE_ID" in os.environ and \
 97                os.environ["ORYX_ENV_TYPE"] == "AppService",
 98        ),
 99    ]
100
101    try:
102      recognized = next(e for e in env if e.check())
103      if recognized is not None:
104        return recognized.name
105    except:
106      return "Unknown"
107
108  def __str__(self):
109    return "driver=python-{0}; runtime=python-{1} env={2}; os={3}".format(
110        self.driverVersion, self.pythonVersion, self.env, self.os).lower()