Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[DPE-5039] - Simple charm ops #19

Merged
merged 11 commits into from
Aug 12, 2024
152 changes: 152 additions & 0 deletions lib/charms/mongodb/v0/mongodb_secrets.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,152 @@
"""Secrets related helper classes/functions."""

# Copyright 2023 Canonical Ltd.
# See LICENSE file for licensing details.

from typing import Dict, Optional

from ops import Secret, SecretInfo
from ops.charm import CharmBase
from ops.model import ModelError, SecretNotFoundError

from config import Config
from exceptions import SecretAlreadyExistsError

# The unique Charmhub library identifier, never change it

# The unique Charmhub library identifier, never change it
LIBID = "89cefc863fd747d7ace12cb508e7bec2"

# Increment this major API version when introducing breaking changes
LIBAPI = 0

# Increment this PATCH version before using `charmcraft publish-lib` or reset
# to 0 if you are raising the major API version
LIBPATCH = 2

APP_SCOPE = Config.Relations.APP_SCOPE
UNIT_SCOPE = Config.Relations.UNIT_SCOPE
Scopes = Config.Relations.Scopes


def generate_secret_label(charm: CharmBase, scope: Scopes) -> str:
"""Generate unique group_mappings for secrets within a relation context.

Defined as a standalone function, as the choice on secret labels definition belongs to the
Application Logic. To be kept separate from classes below, which are simply to provide a
(smart) abstraction layer above Juju Secrets.
"""
members = [charm.app.name, scope]
return f"{'.'.join(members)}"


# Secret cache


class CachedSecret:
"""Abstraction layer above direct Juju access with caching.

The data structure is precisely re-using/simulating Juju Secrets behavior, while
also making sure not to fetch a secret multiple times within the same event scope.
"""

def __init__(self, charm: CharmBase, label: str, secret_uri: Optional[str] = None):
self._secret_meta = None
self._secret_content = {}
self._secret_uri = secret_uri
self.label = label
self.charm = charm

def add_secret(self, content: Dict[str, str], scope: Scopes) -> Secret:
"""Create a new secret."""
if self._secret_uri:
raise SecretAlreadyExistsError(
"Secret is already defined with uri %s", self._secret_uri
)

if scope == Config.Relations.APP_SCOPE:
secret = self.charm.app.add_secret(content, label=self.label)
else:
secret = self.charm.unit.add_secret(content, label=self.label)
self._secret_uri = secret.id
self._secret_meta = secret
return self._secret_meta

@property
def meta(self) -> Optional[Secret]:
"""Getting cached secret meta-information."""
if self._secret_meta:
return self._secret_meta

if not (self._secret_uri or self.label):
return

try:
self._secret_meta = self.charm.model.get_secret(label=self.label)
except SecretNotFoundError:
if self._secret_uri:
self._secret_meta = self.charm.model.get_secret(
id=self._secret_uri, label=self.label
)
return self._secret_meta

def get_content(self) -> Dict[str, str]:
"""Getting cached secret content."""
if not self._secret_content:
if self.meta:
try:
self._secret_content = self.meta.get_content(refresh=True)
except (ValueError, ModelError) as err:
# https://bugs.launchpad.net/juju/+bug/2042596
# Only triggered when 'refresh' is set
known_model_errors = [
"ERROR either URI or label should be used for getting an owned secret but not both",
"ERROR secret owner cannot use --refresh",
]
if isinstance(err, ModelError) and not any(
msg in str(err) for msg in known_model_errors
):
raise
# Due to: ValueError: Secret owner cannot use refresh=True
self._secret_content = self.meta.get_content()
return self._secret_content

def set_content(self, content: Dict[str, str]) -> None:
"""Setting cached secret content."""
if self.meta:
self.meta.set_content(content)
self._secret_content = content

def get_info(self) -> Optional[SecretInfo]:
"""Wrapper function for get the corresponding call on the Secret object if any."""
if self.meta:
return self.meta.get_info()


class SecretCache:
"""A data structure storing CachedSecret objects."""

def __init__(self, charm):
self.charm = charm
self._secrets: Dict[str, CachedSecret] = {}

def get(self, label: str, uri: Optional[str] = None) -> Optional[CachedSecret]:
"""Getting a secret from Juju Secret store or cache."""
if not self._secrets.get(label):
secret = CachedSecret(self.charm, label, uri)
if secret.meta:
self._secrets[label] = secret
return self._secrets.get(label)

def add(self, label: str, content: Dict[str, str], scope: Scopes) -> CachedSecret:
"""Adding a secret to Juju Secret."""
if self._secrets.get(label):
raise SecretAlreadyExistsError(f"Secret {label} already exists")

secret = CachedSecret(self.charm, label)
secret.add_secret(content, scope)
self._secrets[label] = secret
return self._secrets[label]


# END: Secret cache
Loading
Loading