Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Empty file.
8 changes: 8 additions & 0 deletions aws_doc_sdk_examples_tools/labels/known_labels.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
LANGUAGE = "Language"
VERSION = "Version"
SDK = "SDK"
SERVICE = "Service"
ACTION = "Action"
CATEGORY = "Category"
BUNDLE = "Bundle"
CAVEAT = "Caveat"
121 changes: 121 additions & 0 deletions aws_doc_sdk_examples_tools/labels/labels.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Iterable, Iterator, List, Optional


from . import known_labels


@dataclass(frozen=True)
class Label:
name: str
value: str


class LabelSet(defaultdict):
def __init__(self, labels: Iterable[Label]):
super(LabelSet, self).__init__(list)
for label in labels:
self[label.name].append(label.value)

def covers(self, other: "LabelSet") -> bool:
"""covered means all other must be present in this item's Labels, and all values must match exactly."""
try:
for name, value in other.items():
v = self[name]
assert v == value
return True
except AssertionError as _e:
print(_e)
return False
# return all(self[k] == v for k, v in other.items())

def __iter__(self) -> Iterator[Label]:
for k, vs in self.items():
for v in vs:
yield Label(name=k, value=v)


@dataclass
class Labeled:
labels: Iterable[Label] = field(default_factory=list)
_label_set: LabelSet = field(init=False)

def __post_init__(self):
self._label_set = LabelSet(self.labels)

def covers(self, other: "Labeled"):
return self._label_set.covers(other._label_set)


@dataclass
class Snippet(Labeled):
id: str = ""

def __eq__(self, other):
return self.id == other.id

def __hash__(self):
return hash((self.id))


@dataclass
class Sdk(Labeled):
language: str = ""
version: str = ""

def __post_init__(self):
self.labels = [Label(name=known_labels.SDK, value=f"{self.language}:{self.version}")]
self._label_set = LabelSet(self.labels)

def __eq__(self, other):
return self.language == other.language and self.version == other.version

def __hash__(self):
return hash((self.language, self.version))


@dataclass
class Service(Labeled):
name: str = ""
long: str = ""
short: str = ""
sort: str = ""
version: str = ""
# expanded: Optional[ServiceExpanded] = None
api_ref: Optional[str] = None
blurb: Optional[str] = None
# guide: Optional[ServiceGuide] = None

def __eq__(self, other):
return self.name == other.name

def __hash__(self):
return hash((self.name))


@dataclass
class Example(Labeled):
id: str = ""
title: Optional[str] = ""
title_abbrev: Optional[str] = ""
synopsis: Optional[str] = ""
category: Optional[str] = None
# guide_topic: Optional[Url] = None
# doc_filenames: Optional[DocFilenames] = None
synopsis_list: List[str] = field(default_factory=list)

def __eq__(self, other):
return self.id == other.id

def __hash__(self):
return hash((self.id))


def select(items: Iterable[Labeled], for_labels_in: Labeled):
return_set = set()
for item in items:
if item.covers(for_labels_in):
return_set.add(item)
return frozenset(return_set)
# return frozenset(item for item in items if item.covers(for_labels_in))
72 changes: 72 additions & 0 deletions aws_doc_sdk_examples_tools/labels/labels_test.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
from .labels import Example, Label, LabelSet, Sdk, Service, Snippet, select
from . import known_labels


def test_LabelSet_init():
label_set = LabelSet([Label(name="a", value="1"), Label(name="b", value="2"), Label(name="a", value="11")])
assert label_set['a'] == ["1", "11"]
assert label_set['b'] == ["2"]


def test_LabelSet_iter():
label_set = LabelSet([Label(name="a", value="1"), Label(name="b", value="2"), Label(name="a", value="11")])
labels = [*label_set]
assert labels == [Label(name="a", value="1"), Label(name="a", value="11"), Label(name="b", value="2")]


def test_LabelSet_cover():
ls_snippet = LabelSet([Label(name="sdk", value="rust:1"), Label(name="service", value="ec2"), Label(name="action", value="DescribeInstance")])
ls_sdk_rust = LabelSet([Label(name="sdk", value="rust:1")])
ls_sdk_java = LabelSet([Label(name="sdk", value="java:2")])
ls_example_describeinstance = LabelSet([Label(name="service", value="ec2"), Label(name="action", value="DescribeInstance")])

assert ls_snippet.covers(ls_snippet) is True
assert ls_snippet.covers(ls_sdk_rust) is True
assert ls_snippet.covers(ls_sdk_java) is False
assert ls_snippet.covers(ls_example_describeinstance) is True

assert ls_sdk_rust.covers(ls_snippet) is False
assert ls_sdk_rust.covers(ls_sdk_rust) is True
assert ls_sdk_rust.covers(ls_sdk_java) is False
assert ls_sdk_rust.covers(ls_example_describeinstance) is False

assert ls_sdk_java.covers(ls_snippet) is False
assert ls_sdk_java.covers(ls_sdk_rust) is False
assert ls_sdk_java.covers(ls_sdk_java) is True
assert ls_sdk_java.covers(ls_example_describeinstance) is False

assert ls_example_describeinstance.covers(ls_snippet) is False
assert ls_example_describeinstance.covers(ls_sdk_rust) is False
assert ls_example_describeinstance.covers(ls_sdk_java) is False
assert ls_example_describeinstance.covers(ls_example_describeinstance) is True


def test_select():
snippets = [
Snippet(id="rustv1.ec2.DescribeInstance", labels=[Label(name=known_labels.SDK, value="rust:v1"), Label(name=known_labels.SERVICE, value="ec2"), Label(name=known_labels.ACTION, value="DescribeInstance")]),
Snippet(id="rustv1.ec2.RebootInstance", labels=[Label(name=known_labels.SDK, value="rust:v1"), Label(name=known_labels.SERVICE, value="ec2"), Label(name=known_labels.ACTION, value="RebootInstance")]),
Snippet(id="rustv1.ses.Send", labels=[Label(name=known_labels.SDK, value="rust:v1"), Label(name=known_labels.SERVICE, value="ses"), Label(name=known_labels.ACTION, value="Send")]),
Snippet(id="javav2.ec2.DescribeInstance", labels=[Label(name=known_labels.SDK, value="java:v2"), Label(name=known_labels.SERVICE, value="ec2"), Label(name=known_labels.ACTION, value="DescribeInstance")]),
Snippet(id="javav2.ec2.RebootInstance", labels=[Label(name=known_labels.SDK, value="java:v2"), Label(name=known_labels.SERVICE, value="ec2"), Label(name=known_labels.ACTION, value="RebootInstance")]),
Snippet(id="javav2.ses.Send", labels=[Label(name=known_labels.SDK, value="java:v2"), Label(name=known_labels.SERVICE, value="ses"), Label(name=known_labels.ACTION, value="Send")]),
]

examples = [
Example(id="ec2.describe_instance", title="Describe an &EC2; Instance", labels=[Label(name=known_labels.SERVICE, value="ec2"), Label(name=known_labels.ACTION, value="DescribeInstance")]),
Example(id="ec2.reboot_instance", title="Reboot an &EC2; Instance", labels=[Label(name=known_labels.SERVICE, value="ec2"), Label(name=known_labels.ACTION, value="RebootInstance")]),
Example(id="ses.Send", title="Send a message to an &SES; queue", labels=[Label(name=known_labels.SERVICE, value="ses"), Label(name=known_labels.ACTION, value="RebootInstance")]),
Example(id="s3.PutObject", title="Put an object into &S3; storage", labels=[Label(name=known_labels.SERVICE, value="s3"), Label(name=known_labels.ACTION, value="PutObject")])
]

service = Service(name="ec2", labels=[Label(name=known_labels.SERVICE, value="ec2")])

sdk = Sdk(language="rust", version="v1")

example_snippets = select(snippets, examples[0])
service_examples = select(examples, service)
sdk_snippets = select(snippets, sdk)
# sdk_examples = select_all(examples, sdk_snippets)

assert example_snippets == frozenset([snippets[0], snippets[3]])
assert service_examples == frozenset([examples[0], examples[1]])
assert sdk_snippets == frozenset([snippets[0], snippets[1], snippets[2]])
Loading