|
|
|
|
@ -1,30 +1,36 @@
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
|
|
|
|
from typing import TYPE_CHECKING, Any, AsyncGenerator, Literal
|
|
|
|
|
from typing import TYPE_CHECKING, Any, Literal, override
|
|
|
|
|
|
|
|
|
|
from mipac.abstract.action import AbstractAction
|
|
|
|
|
from mipac.http import Route
|
|
|
|
|
from mipac.http import HTTPClient, Route
|
|
|
|
|
from mipac.models.roles import Role, RoleUser
|
|
|
|
|
from mipac.types.meta import IPolicies
|
|
|
|
|
from mipac.types.roles import IRole, IRoleUser
|
|
|
|
|
from mipac.utils.format import remove_dict_missing
|
|
|
|
|
from mipac.utils.pagination import Pagination
|
|
|
|
|
from mipac.utils.util import MISSING
|
|
|
|
|
|
|
|
|
|
if TYPE_CHECKING:
|
|
|
|
|
from mipac.http import HTTPClient
|
|
|
|
|
from mipac.manager.client import ClientManager
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class AdminRoleModelActions(AbstractAction):
|
|
|
|
|
def __init__(self, role_id: str | None = None, *, session: HTTPClient, client: ClientManager):
|
|
|
|
|
class SharedAdminRoleActions:
|
|
|
|
|
def __init__(self, *, session: HTTPClient, client: ClientManager) -> None:
|
|
|
|
|
self._session: HTTPClient = session
|
|
|
|
|
self._client: ClientManager = client
|
|
|
|
|
self._role_id: str | None = role_id
|
|
|
|
|
|
|
|
|
|
async def delete(self, *, role_id: str) -> bool:
|
|
|
|
|
res: bool = await self._session.request(
|
|
|
|
|
Route("POST", "/api/admin/roles/delete"),
|
|
|
|
|
json={"roleId": role_id},
|
|
|
|
|
)
|
|
|
|
|
return res
|
|
|
|
|
|
|
|
|
|
async def update(
|
|
|
|
|
self,
|
|
|
|
|
name: str,
|
|
|
|
|
description: str,
|
|
|
|
|
role_id: str | None = None,
|
|
|
|
|
color: str | None = None,
|
|
|
|
|
iconUrl: str | None = None,
|
|
|
|
|
target: Literal["manual", "conditional"] = "manual",
|
|
|
|
|
@ -32,150 +38,249 @@ class AdminRoleModelActions(AbstractAction):
|
|
|
|
|
is_public: bool = False,
|
|
|
|
|
is_moderator: bool = False,
|
|
|
|
|
is_administrator: bool = False,
|
|
|
|
|
is_explorable: bool = MISSING,
|
|
|
|
|
as_badge: bool = False,
|
|
|
|
|
can_edit_members_by_moderator: bool = False,
|
|
|
|
|
display_order: int = 0,
|
|
|
|
|
policies: dict[Any, Any] | None = None,
|
|
|
|
|
*,
|
|
|
|
|
role_id: str,
|
|
|
|
|
) -> bool:
|
|
|
|
|
role_id = self._role_id or role_id
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("role_idは必須です")
|
|
|
|
|
body = {
|
|
|
|
|
"roleId": role_id,
|
|
|
|
|
"name": name,
|
|
|
|
|
"description": description,
|
|
|
|
|
"color": color,
|
|
|
|
|
"iconUrl": iconUrl,
|
|
|
|
|
"target": target,
|
|
|
|
|
"condFormula": cond_formula or {},
|
|
|
|
|
"isPublic": is_public,
|
|
|
|
|
"isModerator": is_moderator,
|
|
|
|
|
"isAdministrator": is_administrator,
|
|
|
|
|
"asBadge": as_badge,
|
|
|
|
|
"canEditMembersByModerator": can_edit_members_by_moderator,
|
|
|
|
|
"policies": policies or {},
|
|
|
|
|
}
|
|
|
|
|
body = remove_dict_missing(
|
|
|
|
|
{
|
|
|
|
|
"roleId": role_id,
|
|
|
|
|
"name": name,
|
|
|
|
|
"description": description,
|
|
|
|
|
"color": color,
|
|
|
|
|
"iconUrl": iconUrl,
|
|
|
|
|
"target": target,
|
|
|
|
|
"condFormula": cond_formula or {},
|
|
|
|
|
"isPublic": is_public,
|
|
|
|
|
"isModerator": is_moderator,
|
|
|
|
|
"isAdministrator": is_administrator,
|
|
|
|
|
"isExplorable": is_explorable,
|
|
|
|
|
"asBadge": as_badge,
|
|
|
|
|
"canEditMembersByModerator": can_edit_members_by_moderator,
|
|
|
|
|
"displayOrder": display_order,
|
|
|
|
|
"policies": policies or {},
|
|
|
|
|
}
|
|
|
|
|
)
|
|
|
|
|
res: bool = await self._session.request(
|
|
|
|
|
Route("POST", "/api/admin/roles/update"),
|
|
|
|
|
json=body,
|
|
|
|
|
auth=True,
|
|
|
|
|
lower=True,
|
|
|
|
|
remove_none=False,
|
|
|
|
|
)
|
|
|
|
|
return res
|
|
|
|
|
|
|
|
|
|
async def delete(self, role_id: str | None = None) -> bool:
|
|
|
|
|
role_id = self._role_id or role_id
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("role_idは必須です")
|
|
|
|
|
res: bool = await self._session.request(
|
|
|
|
|
Route("POST", "/api/admin/roles/delete"),
|
|
|
|
|
auth=True,
|
|
|
|
|
json={"roleId": role_id},
|
|
|
|
|
lower=True,
|
|
|
|
|
)
|
|
|
|
|
return res
|
|
|
|
|
|
|
|
|
|
async def assign(
|
|
|
|
|
self, user_id: str, role_id: str | None = None, expires_at: int | None = None
|
|
|
|
|
) -> bool:
|
|
|
|
|
async def assign(self, user_id: str, expires_at: int | None = None, *, role_id: str) -> bool:
|
|
|
|
|
"""指定したユーザーに指定したロールを付与します
|
|
|
|
|
|
|
|
|
|
Parameters
|
|
|
|
|
----------
|
|
|
|
|
role_id : str
|
|
|
|
|
ロールのID
|
|
|
|
|
user_id : str
|
|
|
|
|
ロールを付与する対象のユーザーID
|
|
|
|
|
expires_at : int | None, optional
|
|
|
|
|
いつまでロールを付与するか, by default None
|
|
|
|
|
role_id : str
|
|
|
|
|
ロールのID
|
|
|
|
|
|
|
|
|
|
Returns
|
|
|
|
|
-------
|
|
|
|
|
bool
|
|
|
|
|
成功したか否か
|
|
|
|
|
"""
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("role_idは必須です")
|
|
|
|
|
body = {"roleId": role_id, "userId": user_id, "expiresAt": expires_at}
|
|
|
|
|
res: bool = await self._session.request(
|
|
|
|
|
Route("POST", "/api/admin/roles/assign"), auth=True, json=body
|
|
|
|
|
)
|
|
|
|
|
return res
|
|
|
|
|
|
|
|
|
|
async def unassign(self, user_id: str, role_id: str | None = None) -> bool:
|
|
|
|
|
"""指定したユーザーに指定したロールを付与します
|
|
|
|
|
async def unassign(self, user_id: str, *, role_id: str) -> bool:
|
|
|
|
|
"""指定したユーザーから指定したロールを解除します
|
|
|
|
|
|
|
|
|
|
Parameters
|
|
|
|
|
----------
|
|
|
|
|
user_id : str
|
|
|
|
|
ロールを解除するユーザーID
|
|
|
|
|
role_id : str
|
|
|
|
|
ロールのID
|
|
|
|
|
user_id : str
|
|
|
|
|
ロールを付与する対象のユーザーID
|
|
|
|
|
expires_at : int | None, optional
|
|
|
|
|
いつまでロールを付与するか, by default None
|
|
|
|
|
|
|
|
|
|
Returns
|
|
|
|
|
-------
|
|
|
|
|
bool
|
|
|
|
|
成功したか否か
|
|
|
|
|
"""
|
|
|
|
|
role_id = self._role_id or role_id
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("role_idは必須です")
|
|
|
|
|
body = {"roleId": role_id, "userId": user_id}
|
|
|
|
|
res: bool = await self._session.request(
|
|
|
|
|
Route("POST", "/api/admin/roles/unassign"), auth=True, json=body
|
|
|
|
|
)
|
|
|
|
|
return res
|
|
|
|
|
|
|
|
|
|
async def show(self, role_id: str | None = None) -> Role:
|
|
|
|
|
role_id = self._role_id or role_id
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("role_idは必須です")
|
|
|
|
|
res: IRole = await self._session.request(
|
|
|
|
|
Route("POST", "/api/admin/roles/show"),
|
|
|
|
|
json={"roleId": role_id},
|
|
|
|
|
auth=True,
|
|
|
|
|
lower=True,
|
|
|
|
|
)
|
|
|
|
|
return Role(res, client=self._client)
|
|
|
|
|
|
|
|
|
|
async def get_users(
|
|
|
|
|
self,
|
|
|
|
|
role_id: str | None = None,
|
|
|
|
|
since_id: str | None = None,
|
|
|
|
|
until_id: str | None = None,
|
|
|
|
|
limit: int = 100,
|
|
|
|
|
get_all: bool = False,
|
|
|
|
|
) -> AsyncGenerator[RoleUser, None]:
|
|
|
|
|
role_id = self._role_id or role_id
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("role_idは必須です")
|
|
|
|
|
since_id: str = MISSING,
|
|
|
|
|
until_id: str = MISSING,
|
|
|
|
|
limit: int = MISSING,
|
|
|
|
|
*,
|
|
|
|
|
role_id: str,
|
|
|
|
|
) -> RoleUser:
|
|
|
|
|
body = remove_dict_missing(
|
|
|
|
|
{"limit": limit, "sinceId": since_id, "untilId": until_id, "roleId": role_id}
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
raw_role_user: IRoleUser = await self._session.request(
|
|
|
|
|
Route("POST", "/api/admin/roles/users"), json=body
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
if limit > 100:
|
|
|
|
|
raise ValueError("limitは100以下である必要があります")
|
|
|
|
|
return RoleUser(raw_role_user, client=self._client)
|
|
|
|
|
|
|
|
|
|
if get_all:
|
|
|
|
|
limit = 100
|
|
|
|
|
async def get_all_users(
|
|
|
|
|
self,
|
|
|
|
|
since_id: str = MISSING,
|
|
|
|
|
until_id: str = MISSING,
|
|
|
|
|
limit: int = MISSING,
|
|
|
|
|
*,
|
|
|
|
|
role_id: str,
|
|
|
|
|
):
|
|
|
|
|
"""指定したロールを持つユーザーを全て取得します
|
|
|
|
|
|
|
|
|
|
body = {"limit": limit, "sinceId": since_id, "untilId": until_id, "roleId": role_id}
|
|
|
|
|
Parameters
|
|
|
|
|
----------
|
|
|
|
|
since_id : str, optional
|
|
|
|
|
ページネーションの開始位置, by default MISSING
|
|
|
|
|
until_id : str, optional
|
|
|
|
|
ページネーションの終了位置, by default MISSING
|
|
|
|
|
limit : int, optional
|
|
|
|
|
1ページあたりの取得数, by default MISSING
|
|
|
|
|
role_id : str
|
|
|
|
|
ロールのID
|
|
|
|
|
|
|
|
|
|
Returns
|
|
|
|
|
-------
|
|
|
|
|
RoleUser
|
|
|
|
|
ロールユーザー
|
|
|
|
|
"""
|
|
|
|
|
body = remove_dict_missing(
|
|
|
|
|
{"sinceId": since_id, "untilId": until_id, "limit": limit, "roleId": role_id}
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
pagination = Pagination[IRoleUser](
|
|
|
|
|
self._session, Route("POST", "/api/admin/roles/users"), json=body
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
while True:
|
|
|
|
|
while pagination.is_final is False:
|
|
|
|
|
raw_role_users = await pagination.next()
|
|
|
|
|
for role_user in raw_role_users:
|
|
|
|
|
yield RoleUser(role_user, client=self._client)
|
|
|
|
|
for raw_role_user in raw_role_users:
|
|
|
|
|
yield RoleUser(raw_role_user, client=self._client)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class ClientAdminRoleActions(SharedAdminRoleActions):
|
|
|
|
|
def __init__(self, role_id: str, *, session: HTTPClient, client: ClientManager) -> None:
|
|
|
|
|
super().__init__(session=session, client=client)
|
|
|
|
|
self.__role_id: str = role_id
|
|
|
|
|
|
|
|
|
|
@override
|
|
|
|
|
async def delete(self, *, role_id: str | None = None) -> bool:
|
|
|
|
|
role_id = role_id or self.__role_id
|
|
|
|
|
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("required role_id")
|
|
|
|
|
|
|
|
|
|
return await super().delete(role_id=role_id)
|
|
|
|
|
|
|
|
|
|
@override
|
|
|
|
|
async def update(
|
|
|
|
|
self,
|
|
|
|
|
name: str,
|
|
|
|
|
description: str,
|
|
|
|
|
color: str | None = None,
|
|
|
|
|
iconUrl: str | None = None,
|
|
|
|
|
target: Literal["manual"] | Literal["conditional"] = "manual",
|
|
|
|
|
cond_formula: dict[Any, Any] | None = None,
|
|
|
|
|
is_public: bool = False,
|
|
|
|
|
is_moderator: bool = False,
|
|
|
|
|
is_administrator: bool = False,
|
|
|
|
|
is_explorable: bool = MISSING,
|
|
|
|
|
as_badge: bool = False,
|
|
|
|
|
can_edit_members_by_moderator: bool = False,
|
|
|
|
|
display_order: int = 0,
|
|
|
|
|
policies: dict[Any, Any] | None = None,
|
|
|
|
|
*,
|
|
|
|
|
role_id: str | None = None,
|
|
|
|
|
) -> bool:
|
|
|
|
|
role_id = role_id or self.__role_id
|
|
|
|
|
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("required role_id")
|
|
|
|
|
|
|
|
|
|
if get_all is False or pagination.is_final:
|
|
|
|
|
break
|
|
|
|
|
return await super().update(
|
|
|
|
|
name=name,
|
|
|
|
|
description=description,
|
|
|
|
|
color=color,
|
|
|
|
|
iconUrl=iconUrl,
|
|
|
|
|
target=target,
|
|
|
|
|
cond_formula=cond_formula,
|
|
|
|
|
is_public=is_public,
|
|
|
|
|
is_moderator=is_moderator,
|
|
|
|
|
is_administrator=is_administrator,
|
|
|
|
|
is_explorable=is_explorable,
|
|
|
|
|
as_badge=as_badge,
|
|
|
|
|
can_edit_members_by_moderator=can_edit_members_by_moderator,
|
|
|
|
|
display_order=display_order,
|
|
|
|
|
policies=policies,
|
|
|
|
|
role_id=role_id,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
@override
|
|
|
|
|
async def assign(
|
|
|
|
|
self, user_id: str, expires_at: int | None = None, *, role_id: str | None = None
|
|
|
|
|
) -> bool:
|
|
|
|
|
role_id = role_id or self.__role_id
|
|
|
|
|
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("required role_id")
|
|
|
|
|
|
|
|
|
|
return await super().assign(user_id=user_id, expires_at=expires_at, role_id=role_id)
|
|
|
|
|
|
|
|
|
|
@override
|
|
|
|
|
async def unassign(self, user_id: str, *, role_id: str | None = None) -> bool:
|
|
|
|
|
role_id = role_id or self.__role_id
|
|
|
|
|
|
|
|
|
|
if role_id is None:
|
|
|
|
|
raise ValueError("required role_id")
|
|
|
|
|
|
|
|
|
|
return await super().unassign(user_id=user_id, role_id=role_id)
|
|
|
|
|
|
|
|
|
|
@override
|
|
|
|
|
async def get_users(
|
|
|
|
|
self,
|
|
|
|
|
since_id: str = MISSING,
|
|
|
|
|
until_id: str = MISSING,
|
|
|
|
|
limit: int = MISSING,
|
|
|
|
|
*,
|
|
|
|
|
role_id: str,
|
|
|
|
|
) -> RoleUser:
|
|
|
|
|
return await super().get_users(
|
|
|
|
|
since_id=since_id, until_id=until_id, limit=limit, role_id=role_id
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
@override
|
|
|
|
|
async def get_all_users(
|
|
|
|
|
self,
|
|
|
|
|
since_id: str = MISSING,
|
|
|
|
|
until_id: str = MISSING,
|
|
|
|
|
limit: int = MISSING,
|
|
|
|
|
*,
|
|
|
|
|
role_id: str,
|
|
|
|
|
):
|
|
|
|
|
return super().get_all_users(
|
|
|
|
|
since_id=since_id, until_id=until_id, limit=limit, role_id=role_id
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class AdminRoleActions(AdminRoleModelActions):
|
|
|
|
|
def __init__(self, role_id: str | None = None, *, session: HTTPClient, client: ClientManager):
|
|
|
|
|
super().__init__(role_id=role_id, session=session, client=client)
|
|
|
|
|
class AdminRoleActions(SharedAdminRoleActions):
|
|
|
|
|
def __init__(self, *, session: HTTPClient, client: ClientManager):
|
|
|
|
|
super().__init__(session=session, client=client)
|
|
|
|
|
|
|
|
|
|
async def create(
|
|
|
|
|
self,
|
|
|
|
|
|
yupix
4 months ago