# Copyright 2023 Canonical Ltd.
# Licensed under the Apache V2, see LICENCE file for details.
import logging
from . import model
from .errors import JujuEntityNotFoundError
log = logging.getLogger(__name__)
[docs]class Endpoint:
def __init__(self, model, data):
self.model = model
self.data = data
def __repr__(self):
return "<Endpoint {}:{}>".format(self.data["application-name"], self.name)
@property
def application_name(self):
return self.data["application-name"]
@property
def application(self):
"""Application returns the underlying application model from the state.
If no application is found, then a JujuEntityNotFoundError is raised, in
this scenario it is expected that you disconnect and reconnect to the
model.
"""
app_name = self.data["application-name"]
if app_name in self.model.applications:
return self.model.applications[app_name]
raise JujuEntityNotFoundError(app_name, ["application"])
@property
def name(self):
return self.data["relation"]["name"]
@property
def interface(self):
return self.data["relation"]["interface"]
@property
def role(self):
return self.data["relation"]["role"]
@property
def scope(self):
return self.data["relation"]["scope"]
[docs]class Relation(model.ModelEntity):
def __repr__(self):
return f"<Relation id={self.entity_id} {self.key}>"
@property
def endpoints(self):
return [Endpoint(self.model, data) for data in self.safe_data["endpoints"]]
@property
def provides(self):
"""The endpoint on the provides side of this relation, or None."""
for endpoint in self.endpoints:
if endpoint.role == "provider":
return endpoint
return None
@property
def requires(self):
"""The endpoint on the requires side of this relation, or None."""
for endpoint in self.endpoints:
if endpoint.role == "requirer":
return endpoint
return None
@property
def peers(self):
"""The peers endpoint of this relation, or None."""
for endpoint in self.endpoints:
if endpoint.role == "peer":
return endpoint
return None
@property
def is_subordinate(self):
return any(ep.scope == "container" for ep in self.endpoints)
@property
def is_peer(self):
return any(ep.role == "peer" for ep in self.endpoints)
[docs] def matches(self, *specs):
"""Check if this relation matches relationship specs.
Relation specs are strings that would be given to Juju to establish a
relation, and should be in the form ``<application>[:<endpoint_name>]``
where the ``:<endpoint_name>`` suffix is optional. If the suffix is
omitted, this relation will match on any endpoint as long as the given
application is involved.
In other words, this relation will match a spec if that spec could have
created this relation.
:return: True if all specs match.
"""
# Matches expects that the underlying application exists when it walks
# over the endpoints.
# This isn't directly required, but it validates that the framework
# has all the information available to it, when you walk over all the
# relations.
# The one exception is remote-<uuid> applications aren't real
# applications in the general sense of a application, but are more akin
# to a shadow application.
def model_application_exists(app_name):
model_app_name = None
if app_name in self.model.applications:
model_app_name = self.model.applications[app_name].name
elif app_name in self.model.remote_applications:
model_app_name = self.model.remote_applications[app_name].name
elif app_name in self.model.application_offers:
model_app_name = self.model.application_offers[app_name].name
return model_app_name == app_name
for spec in specs:
if ":" in spec:
app_name, endpoint_name = spec.split(":")
else:
app_name, endpoint_name = spec, None
for endpoint in self.endpoints:
if (
app_name == endpoint.application_name
and model_application_exists(app_name)
and endpoint_name in (endpoint.name, None)
):
# found a match for this spec, so move to next one
break
else:
# no match for this spec
return False
return True
@property
def applications(self):
"""All applications involved in this relation."""
return [ep.application for ep in self.endpoints]