Skip to content

services

Provides functions used to manage secrets.

SecretSession

Holds information about the current secret creation session.

Source code in dapla_team_cli/secrets/services.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
class SecretSession:
    """Holds information about the current secret creation session."""

    def __init__(self, project_id: str, secret_id: str, payload: str, client: SecretManagerServiceClient) -> None:
        """Initialize secret info and encode payload."""
        self.project_id = project_id
        self.secret_id = secret_id
        self.payload = payload.encode("UTF-8")
        self.client = client

    def request_creation(self) -> None:
        """Requests google cloud storage client to create a secret."""
        parent = f"projects/{self.project_id}"

        response = self.client.create_secret(
            request={
                "parent": parent,
                "secret_id": self.secret_id,
                "secret": {"replication": {"user_managed": {"replicas": [{"location": "europe-north1"}]}}},
            }
        )

        print(f"Created secret: {response.name}")

    def add_version(self) -> None:
        """Requests google cloud storage client to create a secret."""
        parent = self.client.secret_path(self.project_id, self.secret_id)

        response = self.client.add_secret_version(
            request={
                "parent": parent,
                "payload": {"data": self.payload},
            }
        )

        print(f"Added secret version: {response.name}")

__init__(project_id, secret_id, payload, client)

Initialize secret info and encode payload.

Source code in dapla_team_cli/secrets/services.py 
14
15
16
17
18
19
def __init__(self, project_id: str, secret_id: str, payload: str, client: SecretManagerServiceClient) -> None:
    """Initialize secret info and encode payload."""
    self.project_id = project_id
    self.secret_id = secret_id
    self.payload = payload.encode("UTF-8")
    self.client = client

add_version()

Requests google cloud storage client to create a secret.

Source code in dapla_team_cli/secrets/services.py 
35
36
37
38
39
40
41
42
43
44
45
46
def add_version(self) -> None:
    """Requests google cloud storage client to create a secret."""
    parent = self.client.secret_path(self.project_id, self.secret_id)

    response = self.client.add_secret_version(
        request={
            "parent": parent,
            "payload": {"data": self.payload},
        }
    )

    print(f"Added secret version: {response.name}")

request_creation()

Requests google cloud storage client to create a secret.

Source code in dapla_team_cli/secrets/services.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
def request_creation(self) -> None:
    """Requests google cloud storage client to create a secret."""
    parent = f"projects/{self.project_id}"

    response = self.client.create_secret(
        request={
            "parent": parent,
            "secret_id": self.secret_id,
            "secret": {"replication": {"user_managed": {"replicas": [{"location": "europe-north1"}]}}},
        }
    )

    print(f"Created secret: {response.name}")

get_secret_client()

Get a secret manager seervice client instance.

If in a jupyterhub environment, use HubAuth, otherwise use application default credentials.

Source code in dapla_team_cli/secrets/services.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def get_secret_client() -> SecretManagerServiceClient:
    """Get a secret manager seervice client instance.

    If in a jupyterhub environment, use HubAuth, otherwise use application default credentials.
    """
    if os.getenv("NB_USER") != "jovyan":
        return secretmanager.SecretManagerServiceClient()

    hub = HubAuth()
    response = requests.get(
        os.environ["LOCAL_USER_PATH"],
        headers={"Authorization": "token {hub.api_token}"},
        cert=(str(hub.certfile), str(hub.keyfile)),
        verify=str(hub.client_ca),
        allow_redirects=False,
        timeout=10,
    )

    token = response.json()["exchanged_tokens"]["google"]["access_token"]
    credentials = Credentials(token=token)
    return secretmanager.SecretManagerServiceClient(credentials=credentials)