-
Notifications
You must be signed in to change notification settings - Fork 12
/
pubsub_context.py
33 lines (26 loc) · 1.09 KB
/
pubsub_context.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from base64 import b64encode
from typing import Mapping, Optional, NamedTuple
from absl import logging
import pkg_resources
from google.protobuf import struct_pb2
class _Semver(NamedTuple):
major: int
minor: int
def _version() -> _Semver:
version = pkg_resources.get_distribution("google-cloud-pubsublite").version
splits = version.split(".")
if len(splits) != 3:
logging.info(f"Failed to extract semver from {version}.")
return _Semver(0, 0)
return _Semver(int(splits[0]), int(splits[1]))
def pubsub_context(framework: Optional[str] = None) -> Mapping[str, str]:
"""Construct the pubsub context mapping for the given framework."""
context = struct_pb2.Struct()
context.fields["language"].string_value = "PYTHON"
if framework:
context.fields["framework"].string_value = framework
version = _version()
context.fields["major_version"].number_value = version.major
context.fields["minor_version"].number_value = version.minor
encoded = b64encode(context.SerializeToString()).decode("utf-8")
return {"x-goog-pubsub-context": encoded}