Skip to content

Commit

Permalink
chore: add more docstrings
Browse files Browse the repository at this point in the history
  • Loading branch information
anitarua committed Sep 23, 2024
1 parent d540a98 commit 44addc9
Show file tree
Hide file tree
Showing 6 changed files with 214 additions and 0 deletions.
16 changes: 16 additions & 0 deletions src/momento/auth/access_control/disposable_token_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,21 +13,29 @@

@dataclass
class AllCacheItems:
"""Indicates permission to access all items in a cache."""

pass


@dataclass
class CacheItemKey:
"""The key of a cache item."""

key: str


@dataclass
class CacheItemKeyPrefix:
"""The prefix of a cache item key."""

key_prefix: str


@dataclass
class CacheItemSelector:
"""A selection of cache items to grant permissions to, either all cache items, a specific cache item, or a items that match a specified key prefix."""

cache_item: Union[CacheItemKey, CacheItemKeyPrefix, AllCacheItems, str]

def is_all_cache_items(self) -> bool:
Expand All @@ -36,19 +44,27 @@ def is_all_cache_items(self) -> bool:

@dataclass
class DisposableTokenCachePermission(CachePermission):
"""Encapsulates the information needed to grant permissions to a cache item."""

cache_item_selector: CacheItemSelector


@dataclass
class DisposableTokenCachePermissions:
"""A list of permissions to grant to a disposable token."""

disposable_token_permissions: List[DisposableTokenCachePermission]


@dataclass
class DisposableTokenScope:
"""A set of permissions to grant to a disposable token."""

disposable_token_scope: Union[Permissions, DisposableTokenCachePermissions]


@dataclass
class DisposableTokenProps:
"""Additional properties for a disposable token, such as token_id, which can be used to identify the source of a token."""

token_id: Optional[str]
110 changes: 110 additions & 0 deletions src/momento/auth/access_control/disposable_token_scopes.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,24 @@


class DisposableTokenScopes:
"""Disposable Token Scopes.
Convenience methods for creating permission scopes for disposable tokens.
"""

@staticmethod
def cache_key_read_write(
cache: Union[AllCaches, CacheName, str], key: Union[CacheItemKey, str]
) -> DisposableTokenScope:
"""Create permissions for read-write access to a specific key in specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
key (Union[CacheItemKey, str]): The key to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
_key = key if isinstance(key, CacheItemKey) else CacheItemKey(key)
scope = DisposableTokenCachePermissions(
[
Expand All @@ -48,6 +62,15 @@ def cache_key_read_write(
def cache_key_prefix_read_write(
cache: Union[AllCaches, CacheName, str], key_prefix: Union[CacheItemKeyPrefix, str]
) -> DisposableTokenScope:
"""Create permissions for read-write access to keys that match a specific key prefix in specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
key_prefix (Union[CacheItemKey, str]): The key prefix to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
_prefix = key_prefix if isinstance(key_prefix, CacheItemKeyPrefix) else CacheItemKeyPrefix(key_prefix)
scope = DisposableTokenCachePermissions(
[
Expand All @@ -64,6 +87,15 @@ def cache_key_prefix_read_write(
def cache_key_read_only(
cache: Union[AllCaches, CacheName, str], key: Union[CacheItemKey, str]
) -> DisposableTokenScope:
"""Create permissions for read-only access to a specific key in specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
key (Union[CacheItemKey, str]): The key to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
_key = key if isinstance(key, CacheItemKey) else CacheItemKey(key)
scope = DisposableTokenCachePermissions(
[
Expand All @@ -80,6 +112,15 @@ def cache_key_read_only(
def cache_key_prefix_read_only(
cache: Union[AllCaches, CacheName, str], key_prefix: Union[CacheItemKeyPrefix, str]
) -> DisposableTokenScope:
"""Create permissions for read-only access to keys that match a specific key prefix in specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
key_prefix (Union[CacheItemKey, str]): The key prefix to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
_prefix = key_prefix if isinstance(key_prefix, CacheItemKeyPrefix) else CacheItemKeyPrefix(key_prefix)
scope = DisposableTokenCachePermissions(
[
Expand All @@ -96,6 +137,15 @@ def cache_key_prefix_read_only(
def cache_key_write_only(
cache: Union[AllCaches, CacheName, str], key: Union[CacheItemKey, str]
) -> DisposableTokenScope:
"""Create permissions for write-only access to a specific key in specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
key (Union[CacheItemKey, str]): The key to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
_key = key if isinstance(key, CacheItemKey) else CacheItemKey(key)
scope = DisposableTokenCachePermissions(
[
Expand All @@ -112,6 +162,15 @@ def cache_key_write_only(
def cache_key_prefix_write_only(
cache: Union[AllCaches, CacheName, str], key_prefix: Union[CacheItemKeyPrefix, str]
) -> DisposableTokenScope:
"""Create permissions for write-only access to keys that match a specific key prefix in specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
key_prefix (Union[CacheItemKey, str]): The key prefix to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
_prefix = key_prefix if isinstance(key_prefix, CacheItemKeyPrefix) else CacheItemKeyPrefix(key_prefix)
scope = DisposableTokenCachePermissions(
[
Expand All @@ -126,6 +185,14 @@ def cache_key_prefix_write_only(

@staticmethod
def cache_read_write(cache: Union[AllCaches, CacheName, str]) -> DisposableTokenScope:
"""Create permissions for read-write access to specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
scope = Permissions(
[
CachePermission(
Expand All @@ -138,6 +205,14 @@ def cache_read_write(cache: Union[AllCaches, CacheName, str]) -> DisposableToken

@staticmethod
def cache_read_only(cache: Union[AllCaches, CacheName, str]) -> DisposableTokenScope:
"""Create permissions for read-only access to specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
scope = Permissions(
[
CachePermission(
Expand All @@ -150,6 +225,14 @@ def cache_read_only(cache: Union[AllCaches, CacheName, str]) -> DisposableTokenS

@staticmethod
def cache_write_only(cache: Union[AllCaches, CacheName, str]) -> DisposableTokenScope:
"""Create permissions for write-only access to specified cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
scope = Permissions(
[
CachePermission(
Expand All @@ -164,6 +247,15 @@ def cache_write_only(cache: Union[AllCaches, CacheName, str]) -> DisposableToken
def topic_publish_subscribe(
cache: Union[AllCaches, CacheName, str], topic: Union[TopicName, AllTopics, str]
) -> DisposableTokenScope:
"""Create permissions for publish-subscribe access to specified topic(s) and cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
topic (Union[TopicName, AllTopics, str]): The topic(s) to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
scope = Permissions(
[
TopicPermission(
Expand All @@ -179,6 +271,15 @@ def topic_publish_subscribe(
def topic_subscribe_only(
cache: Union[AllCaches, CacheName, str], topic: Union[TopicName, AllTopics, str]
) -> DisposableTokenScope:
"""Create permissions for subscribe-only access to specified topic(s) and cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
topic (Union[TopicName, AllTopics, str]): The topic(s) to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
scope = Permissions(
[
TopicPermission(
Expand All @@ -194,6 +295,15 @@ def topic_subscribe_only(
def topic_publish_only(
cache: Union[AllCaches, CacheName, str], topic: Union[TopicName, AllTopics, str]
) -> DisposableTokenScope:
"""Create permissions for publish-only access to specified topic(s) and cache(s).
Args:
cache (Union[AllCaches, CacheName, str]): The cache(s) to grant permission to.
topic (Union[TopicName, AllTopics, str]): The topic(s) to grant permission to.
Returns:
DisposableTokenScope: A set of permissions to grant to a disposable token.
"""
scope = Permissions(
[
TopicPermission(
Expand Down
27 changes: 27 additions & 0 deletions src/momento/auth/access_control/permission_scope.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,57 +9,78 @@

@dataclass
class AllCaches:
"""Indicates permission to access all caches."""

pass


@dataclass
class AllTopics:
"""Indicates permission to access all topics."""

pass


@dataclass
class PredefinedScope:
"""Indicates a predefined permission scope."""

pass


class CacheRole(Enum):
"""The permission level for a cache."""

READ_WRITE = "read_write"
READ_ONLY = "read_only"
WRITE_ONLY = "write_only"


@dataclass
class CacheName:
"""The name of a cache."""

name: str


@dataclass
class CacheSelector:
"""A selection of caches to grant permissions to, either all caches or a specific cache."""

cache: Union[CacheName, AllCaches, str]

def is_all_caches(self) -> bool:
"""Check if the cache selector is for all caches."""
return isinstance(self.cache, AllCaches)


@dataclass
class CachePermission:
"""Encapsulates the information needed to grant permissions to a cache."""

cache_selector: CacheSelector
role: CacheRole


class TopicRole(Enum):
"""The permission level for a topic."""

PUBLISH_SUBSCRIBE = "publish_subscribe"
SUBSCRIBE_ONLY = "subscribe_only"
PUBLISH_ONLY = "publish_only"


@dataclass
class TopicName:
"""The name of a topic."""

name: str


@dataclass
class TopicSelector:
"""A selection of topics to grant permissions to, either all topics or a specific topic."""

topic: Union[TopicName, AllTopics, str]

def is_all_topics(self) -> bool:
Expand All @@ -68,6 +89,8 @@ def is_all_topics(self) -> bool:

@dataclass
class TopicPermission:
"""Encapsulates the information needed to grant permissions to a topic."""

role: TopicRole
cache_selector: CacheSelector
topic_selector: TopicSelector
Expand All @@ -78,6 +101,8 @@ class TopicPermission:

@dataclass
class Permissions:
"""A list of permissions to grant to an API key."""

permissions: List[Permission]


Expand All @@ -95,6 +120,8 @@ class Permissions:

@dataclass
class PermissionScope:
"""A set of permissions to grant to an API key, either a predefined scope or a custom scope."""

permission_scope: Union[Permissions, PredefinedScope]

def is_all_data_read_write(self) -> bool:
Expand Down
Loading

0 comments on commit 44addc9

Please sign in to comment.