Skip to content

Commit

Permalink
Typed model.interaction
Browse files Browse the repository at this point in the history
  • Loading branch information
eunwoo1104 committed Sep 10, 2021
1 parent d3664b0 commit 7512061
Show file tree
Hide file tree
Showing 8 changed files with 172 additions and 163 deletions.
2 changes: 1 addition & 1 deletion dico/base/http.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ class HTTPRequestBase(ABC):
This module isn't intended to be directly used. It is recommended to request via APIClient.
"""

BASE_URL = "https://discord.com/api/v9"
BASE_URL: str = "https://discord.com/api/v9"

@abstractmethod
def request(self, route: str, meth: str, body: typing.Any = None, *, is_json: bool = False, reason_header: str = None, retry: int = 3, **kwargs):
Expand Down
18 changes: 9 additions & 9 deletions dico/base/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ def __init__(self, client, resp, **kwargs):
self.id: Snowflake = Snowflake(resp["id"])
self.client: APIClient = client

def __int__(self):
def __int__(self) -> int:
return int(self.id)

def update(self, new_resp, **kwargs):
Expand Down Expand Up @@ -87,9 +87,9 @@ def __setattr__(self, key, value):


class FlagBase:
def __init__(self, *args, **kwargs):
self.values = {x: getattr(self, x) for x in dir(self) if isinstance(getattr(self, x), int)}
self.value = 0
def __init__(self, *args: str, **kwargs: bool):
self.values: typing.Dict[str, int] = {x: getattr(self, x) for x in dir(self) if isinstance(getattr(self, x), int)}
self.value: int = 0
for x in args:
if x.upper() not in self.values:
raise AttributeError(f"invalid name: `{x}`")
Expand All @@ -111,7 +111,7 @@ def __iter__(self):
if self.has(k):
yield v

def has(self, name: str):
def has(self, name: str) -> bool:
if name.upper() not in self.values:
raise AttributeError(f"invalid name: `{name}`")
return (self.value & self.values[name.upper()]) == self.values[name.upper()]
Expand Down Expand Up @@ -144,8 +144,8 @@ def from_value(cls, value: int):

class TypeBase:
def __init__(self, value):
self.values = {getattr(self, x): x for x in dir(self) if isinstance(getattr(self, x), int)}
self.value = value
self.values: typing.Dict[int, str] = {getattr(self, x): x for x in dir(self) if isinstance(getattr(self, x), int)}
self.value: int = value

if self.value not in self.values:
raise AttributeError(f"invalid value: `{value}`")
Expand All @@ -159,13 +159,13 @@ def __int__(self):
def __getattr__(self, item):
return self.is_type(item)

def is_type(self, name: str):
def is_type(self, name: str) -> bool:
values = {y: x for x, y in self.values.items()}
if name.upper() not in values:
raise AttributeError(f"invalid name: `{name}`")
return self.value == values[name.upper()]

@classmethod
def to_string(cls, value):
def to_string(cls, value) -> str:
values = {getattr(cls, x): x for x in dir(cls) if isinstance(getattr(cls, x), int)}
return values.get(value)
24 changes: 12 additions & 12 deletions dico/http/async_http.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,19 +12,19 @@
class AsyncHTTPRequest(HTTPRequestBase):
def __init__(self,
token: str,
loop: asyncio.AbstractEventLoop = None,
session: aiohttp.ClientSession = None,
default_retry: int = 3):
self.token = token.lstrip("Bot ")
self.logger = logging.getLogger("dico.http")
self.loop = loop or asyncio.get_event_loop()
self.session = session or aiohttp.ClientSession(loop=self.loop)
self.default_retry = default_retry
self._close_on_del = False
loop: typing.Optional[asyncio.AbstractEventLoop] = None,
session: typing.Optional[aiohttp.ClientSession] = None,
default_retry: typing.Optional[int] = 3):
self.token: str = token.lstrip("Bot ")
self.logger: logging.Logger = logging.getLogger("dico.http")
self.loop: asyncio.AbstractEventLoop = loop or asyncio.get_event_loop()
self.session: aiohttp.ClientSession = session or aiohttp.ClientSession(loop=self.loop)
self.default_retry: int = default_retry
self._close_on_del: bool = False
if not session:
self._close_on_del = True
self._closed = False
self.ratelimits = RatelimitHandler()
self._closed: bool = False
self.ratelimits: RatelimitHandler = RatelimitHandler()

def __del__(self):
if not self._closed:
Expand Down Expand Up @@ -240,7 +240,7 @@ def edit_webhook_message(self,
form.add_field(name, f, filename=sel.name, content_type="application/octet-stream")
return self.request(f"/webhooks/{webhook_id}/{webhook_token}/messages/{message_id}", "PATCH", form)

async def download(self, url):
async def download(self, url) -> bytes:
async with self.session.get(url) as resp:
if resp.status == 200:
return await resp.read()
Expand Down
5 changes: 3 additions & 2 deletions dico/http/ratelimit.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import asyncio
import datetime
import typing


class __EmptyLocker:
Expand All @@ -15,7 +16,7 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):

class RatelimitHandler:
def __init__(self):
self.lockers = {}
self.lockers: typing.Dict[str, typing.Optional[dict]] = {}
self.buckets = {}
self.global_locker = asyncio.Lock()

Expand All @@ -27,7 +28,7 @@ def to_locker_key(meth, route):
def utc(self):
return datetime.datetime.utcnow()

def get_locker(self, meth, route):
def get_locker(self, meth, route) -> dict:
locker_key = self.to_locker_key(meth, route)
if locker_key not in self.lockers:
self.lockers[locker_key] = None
Expand Down
5 changes: 4 additions & 1 deletion dico/model/application.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,13 +4,16 @@
from ..base.model import FlagBase, TypeBase
from ..utils import cdn_url

if typing.TYPE_CHECKING:
from ..api import APIClient


class Application:
TYPING = typing.Union[int, str, Snowflake, "Application"]
RESPONSE = typing.Union["Application", typing.Awaitable["Application"]]

def __init__(self, client, resp):
self.client = client
self.client: typing.Type[APIClient] = client
self.id = Snowflake(resp["id"])
self.name = resp["name"]
self.icon = resp["icon"]
Expand Down
80 changes: 40 additions & 40 deletions dico/model/interactions/commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,24 +13,24 @@ class ApplicationCommand:
def __init__(self,
name: str,
description: str,
command_type: typing.Union[int, "ApplicationCommandTypes"] = 1,
options: typing.List["ApplicationCommandOption"] = None,
default_permission: bool = True,
command_type: typing.Optional[typing.Union[int, "ApplicationCommandTypes"]] = 1,
options: typing.Optional[typing.List["ApplicationCommandOption"]] = None,
default_permission: typing.Optional[bool] = True,
**resp):
self.id = Snowflake.optional(resp.get("id"))
self.type = ApplicationCommandTypes(int(command_type))
self.application_id = Snowflake.optional(resp.get("application_id"))
self.name = name
self.description = description
self.options = options or []
self.default_permission = default_permission
self.id: typing.Optional[Snowflake] = Snowflake.optional(resp.get("id"))
self.type: ApplicationCommandTypes = ApplicationCommandTypes(int(command_type))
self.application_id: typing.Optional[Snowflake] = Snowflake.optional(resp.get("application_id"))
self.name: str = name
self.description: str = description
self.options: typing.List[ApplicationCommandOption] = options or []
self.default_permission: bool = default_permission
self.__command_creation = not resp

def __int__(self):
def __int__(self) -> int:
if self.id:
return int(self.id)

def to_dict(self):
def to_dict(self) -> dict:
resp = {"name": self.name, "description": self.description, "options": [x.to_dict() for x in self.options], "default_permission": self.default_permission, "type": int(self.type)}
if self.__command_creation:
return resp
Expand All @@ -56,18 +56,18 @@ def __init__(self,
option_type: typing.Union["ApplicationCommandOptionType", int],
name: str,
description: str,
required: bool = False,
choices: typing.List["ApplicationCommandOptionChoice"] = None,
options: typing.List["ApplicationCommandOption"] = None,
required: typing.Optional[bool] = False,
choices: typing.Optional[typing.List["ApplicationCommandOptionChoice"]] = None,
options: typing.Optional[typing.List["ApplicationCommandOption"]] = None,
**kw):
self.type = ApplicationCommandOptionType(int(option_type))
self.name = name
self.description = description
self.required = required
self.choices = choices or []
self.options = options or []

def to_dict(self):
self.type: ApplicationCommandOptionType = ApplicationCommandOptionType(int(option_type))
self.name: str = name
self.description: str = description
self.required: bool = required
self.choices: typing.List[ApplicationCommandOptionChoice] = choices or []
self.options: typing.List[ApplicationCommandOption] = options or []

def to_dict(self) -> dict:
ret = {"type": int(self.type), "name": self.name, "description": self.description, "required": self.required,
"choices": [x.to_dict() for x in self.choices], "options": [x.to_dict() for x in self.options]}
if not ret["options"]:
Expand Down Expand Up @@ -99,10 +99,10 @@ class ApplicationCommandOptionType(TypeBase):

class ApplicationCommandOptionChoice:
def __init__(self, name: str, value: typing.Union[str, int, float], **kw):
self.name = name
self.value = value
self.name: str = name
self.value: typing.Union[str, int, float] = value

def to_dict(self):
def to_dict(self) -> dict:
return {"name": self.name, "value": self.value}

@classmethod
Expand All @@ -115,23 +115,23 @@ class GuildApplicationCommandPermissions:
RESPONSE_AS_LIST = typing.Union[typing.List["GuildApplicationCommandPermissions"], typing.Awaitable[typing.List["GuildApplicationCommandPermissions"]]]

def __init__(self, resp: dict):
self.id = Snowflake(resp["id"])
self.application_id = Snowflake(resp["application_id"])
self.guild_id = Snowflake(resp["guild_id"])
self.permissions = [ApplicationCommandPermissions.create(x) for x in resp["permissions"]]
self.id: Snowflake = Snowflake(resp["id"])
self.application_id: Snowflake = Snowflake(resp["application_id"])
self.guild_id: Snowflake = Snowflake(resp["guild_id"])
self.permissions: typing.List[ApplicationCommandPermissions] = [ApplicationCommandPermissions.create(x) for x in resp["permissions"]]


class ApplicationCommandPermissions:
def __init__(self,
target: typing.Union[int, Role, User],
target: typing.Union[Role.TYPING, User.TYPING],
permission_type: typing.Union[int, "ApplicationCommandPermissionType"],
permission: bool,
**kw):
self.id = Snowflake.ensure_snowflake(int(target))
self.type = ApplicationCommandPermissionType(int(permission_type))
self.permission = permission
self.id: typing.Optional[Snowflake] = Snowflake.ensure_snowflake(int(target))
self.type: ApplicationCommandPermissionType = ApplicationCommandPermissionType(int(permission_type))
self.permission: bool = permission

def to_dict(self):
def to_dict(self) -> dict:
return {"id": str(self.id), "type": int(self.type), "permission": self.permission}

@classmethod
Expand All @@ -147,8 +147,8 @@ class ApplicationCommandPermissionType(TypeBase):


class ApplicationCommandInteractionDataOption:
def __init__(self, resp):
self.name = resp["name"]
self.type = ApplicationCommandOptionType(resp["type"])
self.value = resp.get("value")
self.options = [ApplicationCommandInteractionDataOption(x) for x in resp.get("options", [])]
def __init__(self, resp: dict):
self.name: str = resp["name"]
self.type: ApplicationCommandOptionType = ApplicationCommandOptionType(resp["type"])
self.value: typing.Optional = resp.get("value")
self.options: typing.List[ApplicationCommandInteractionDataOption] = [ApplicationCommandInteractionDataOption(x) for x in resp.get("options", [])]
Loading

0 comments on commit 7512061

Please sign in to comment.