Skip to content

Commit

Permalink
fix: replace incorrect bucket permission values
Browse files Browse the repository at this point in the history
Existing values will still work, with a deprecation warning.

Co-Authored-By: Tim Holm <[email protected]>
  • Loading branch information
jyecusch and tjholm committed Apr 10, 2024
1 parent 3a5c89b commit 4486752
Showing 1 changed file with 31 additions and 9 deletions.
40 changes: 31 additions & 9 deletions nitric/resources/buckets.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@
from dataclasses import dataclass
from datetime import timedelta
from enum import Enum
from typing import Callable, List, Literal, Optional, Union
from typing import Callable, List, Literal, Optional, Union, cast
from warnings import warn

import betterproto
Expand Down Expand Up @@ -176,8 +176,6 @@ async def download_url(self, expiry: Optional[Union[timedelta, int]] = None):

async def _sign_url(self, mode: FileMode = FileMode.READ, expiry: Optional[Union[timedelta, int]] = None):
"""Generate a signed URL for reading or writing to a file."""
warn("File.sign_url() is deprecated, use upload_url() or download_url() instead", DeprecationWarning)

if expiry is None:
expiry = timedelta(seconds=600)
if not isinstance(expiry, timedelta):
Expand All @@ -194,7 +192,25 @@ async def _sign_url(self, mode: FileMode = FileMode.READ, expiry: Optional[Union
raise exception_from_grpc_error(grpc_err) from grpc_err


BucketPermission = Literal["reading", "writing", "deleting"]
LegacyBucketPermission = Literal["reading", "writing", "deleting"]
BucketPermission = Literal["read", "write", "delete"]

legacy_perms: List[LegacyBucketPermission] = ["reading", "writing", "deleting"]
new_perms: List[BucketPermission] = ["read", "write", "delete"]


def check_permission(permission: Union[LegacyBucketPermission, BucketPermission]) -> BucketPermission:
"""Check if the permission is valid and return the new permission if it is a legacy permission."""
if permission in legacy_perms:
new_perm = new_perms[legacy_perms.index(cast(LegacyBucketPermission, permission))]
warn(
f"The permission '{permission}' is deprecated. Use '{new_perm}' instead.", DeprecationWarning, stacklevel=2
)
return new_perm
elif permission in new_perms:
return cast(BucketPermission, permission)
else:
raise ValueError("Invalid permission value, must be one of 'read', 'write', or 'delete'.")


class BucketNotificationWorkerOptions:
Expand Down Expand Up @@ -233,19 +249,25 @@ async def _register(self) -> None:

def _perms_to_actions(self, *args: BucketPermission) -> List[Action]:
permission_actions_map: dict[BucketPermission, List[Action]] = {
"reading": [Action.BucketFileGet, Action.BucketFileList],
"writing": [Action.BucketFilePut],
"deleting": [Action.BucketFileDelete],
"read": [Action.BucketFileGet, Action.BucketFileList],
"write": [Action.BucketFilePut],
"delete": [Action.BucketFileDelete],
}

return [action for perm in args for action in permission_actions_map[perm]]

def _to_resource_id(self) -> ResourceIdentifier:
return ResourceIdentifier(name=self.name, type=ResourceType.Bucket) # type:ignore

def allow(self, perm: BucketPermission, *args: BucketPermission) -> BucketRef:
def allow(
self,
perm: Union[LegacyBucketPermission, BucketPermission],
*args: Union[LegacyBucketPermission, BucketPermission],
) -> BucketRef:
"""Request the required permissions for this resource."""
str_args = [str(perm)] + [str(permission) for permission in args]
all_perms: List[BucketPermission] = [check_permission(perm)] + [check_permission(p) for p in args]

str_args = [str(permission) for permission in all_perms]
self._register_policy(*str_args)

return BucketRef(self.name)
Expand Down

0 comments on commit 4486752

Please sign in to comment.