Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
35 changes: 35 additions & 0 deletions alpacloud/eztag/BUILD
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
python_sources()

python_tests(
name="tests",
# dependencies=["./test_resources:k8s_objs"],
)

python_test_utils(
name="test_utils",
)

python_distribution(
name="alpacloud.eztag",
repositories=["@alpacloud.eztag"],
dependencies=[":eztag"],
long_description_path="alpacloud/eztag/readme.md",
provides=python_artifact(
name="alpacloud_eztag",
version="0.1.0",
description="A library for filtering things based on tags",
author="Daniel Goldman",
classifiers=[
"Development Status :: 3 - Alpha",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Topic :: Utilities",
"Topic :: System :: Systems Administration",
],
license="Round Robin 2.0.0",
long_description_content_type="text/markdown",
),
)
Empty file added alpacloud/eztag/__init__.py
Empty file.
98 changes: 98 additions & 0 deletions alpacloud/eztag/logic.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,98 @@
"""Expressions for tag filtering"""

from abc import ABC
from dataclasses import dataclass
from typing import Callable

from alpacloud.eztag.tag import TagSet


class Expr(ABC):
"""A predicate"""

def check(self, tags: TagSet) -> bool:
"""Check if the condition is satisfied"""
raise NotImplementedError


@dataclass(frozen=True, slots=True)
class Cond_(Expr):
"""A condition that checks if a tag set satisfies a predicate"""

f: Callable[[TagSet], bool]

def check(self, tags: TagSet) -> bool:
return self.f(tags)


@dataclass(frozen=True, slots=True)
class And_(Expr):
"""AND of multiple conditions"""

conds: list[Cond_]

def check(self, tags: TagSet) -> bool:
return all(cond.check(tags) for cond in self.conds)


@dataclass(frozen=True, slots=True)
class Or_(Expr):
"""OR of multiple conditions"""

conds: list[Cond_]

def check(self, tags: TagSet) -> bool:
return any(cond.check(tags) for cond in self.conds)


@dataclass(frozen=True, slots=True)
class Not_(Expr):
"""Negation of a condition"""

cond: Cond_

def check(self, tags: TagSet) -> bool:
return not self.cond.check(tags)


@dataclass(frozen=True, slots=True)
class TagHas(Expr):
"""Check if a tag set has a given tag, with any value"""

k: str

def check(self, tags: TagSet) -> bool:
return tags.has(self.k)


@dataclass(frozen=True, slots=True)
class TagMatch(Expr):
"""Check if a tag set has a given tag with a specific value"""

k: str
v: str | None

def check(self, tags: TagSet) -> bool:
return tags.match(self.k, self.v)


@dataclass(frozen=True, slots=True)
class TagRematch(Expr):
"""Check if a tag set has a given tag with value matching a regular expression"""

k: str
v: str

def check(self, tags: TagSet) -> bool:
return tags.rematch(self.k, self.v)


@dataclass(frozen=True, slots=True)
class TagContains(Expr):
"""Check if a tag set has a given tag with a specific value"""

k: str
v: str

def check(self, tags: TagSet) -> bool:
return tags.contains(self.k, self.v)
57 changes: 57 additions & 0 deletions alpacloud/eztag/multidict.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
"""Generic multidict implementation. A multidict allows multiple values for the same key."""

from __future__ import annotations

from typing import Iterable, TypeAlias, TypeGuard

K: TypeAlias = str
V: TypeAlias = str | None


def _is_collection(obj) -> TypeGuard[Iterable]:
"""
Checks if an object is an iterable collection, excluding strings and bytes.
"""
return isinstance(obj, Iterable) and not isinstance(obj, (str, bytes, bytearray))


class MultiDict:
"""
A dictionary that allows multiple values for the same key.
This allows us to have a tag set like `env=prd, env=stg`
"""

def __init__(self):
self.d: dict[K, set[V]] = {}

@classmethod
def from_dict(cls, d: dict[K, V]) -> MultiDict:
"""Create a multidict from a dict of key-value pairs"""
md = cls()
for k, v in d.items():
md[k] = {v}
return md

@classmethod
def create(cls, d: dict[K, Iterable[V] | V]) -> MultiDict:
"""
Create a multidict from a dict of key-value pairs or key-list of values pairs
"""
md = cls()
for k, vs in d.items():
n: set[V]
if not _is_collection(vs):
n = {vs} # type: ignore # idk typeguard
else:
n = set(vs)
md.d[k] = n
return md

def __getitem__(self, key) -> set[V]:
return self.d[key]

def __setitem__(self, key, value):
self.d.setdefault(key, set()).add(value)

def __contains__(self, key) -> bool:
return key in self.d
Loading
Loading