yawaflua/Flask-Discord
1
0
Fork 0
mirror of https://github.com/yawaflua/Flask-Discord.git synced 2026-02-04 10:14:15 +02:00
Code Issues Packages Projects Releases Wiki Activity

replaced all flask references with quart references

Browse Source
This commit is contained in:
Philip Dowie
2020-08-15 23:27:15 +12:00
parent d23441a070
commit 2076f1c1df
21 changed files with 120 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
quart_discord/__init__.py Normal file
View File

@@ -0,0 +1,17 @@
from .exceptions import *
from .utils import *
from .client import DiscordOAuth2Session
__all__ = [
"DiscordOAuth2Session",
"requires_authorization",
"HttpException",
"RateLimited",
"Unauthorized",
]
__version__ = "0.1.60"

178
quart_discord/_http.py Normal file
View File

@@ -0,0 +1,178 @@
import cachetools
import requests
import typing
import json
import os
import abc
from . import configs
from . import exceptions
from quart import session, request
from collections.abc import Mapping
from requests_oauthlib import OAuth2Session
class DiscordOAuth2HttpClient(abc.ABC):
"""An OAuth2 http abstract base class providing some factory methods.
This class is meant to be overridden by :py:class:`quart_discord.DiscordOAuth2Session` and should not be
used directly.
"""
SESSION_KEYS = [
"DISCORD_USER_ID",
"DISCORD_OAUTH2_STATE",
"DISCORD_OAUTH2_TOKEN",
]
def __init__(self, app, client_id=None, client_secret=None, redirect_uri=None, bot_token=None, users_cache=None):
self.client_id = client_id or app.config["DISCORD_CLIENT_ID"]
self.__client_secret = client_secret or app.config["DISCORD_CLIENT_SECRET"]
self.redirect_uri = redirect_uri or app.config["DISCORD_REDIRECT_URI"]
self.__bot_token = bot_token or app.config.get("DISCORD_BOT_TOKEN", str())
self.users_cache = cachetools.LFUCache(
app.config.get("DISCORD_USERS_CACHE_MAX_LIMIT", configs.DISCORD_USERS_CACHE_DEFAULT_MAX_LIMIT)
) if users_cache is None else users_cache
if not issubclass(self.users_cache.__class__, Mapping):
raise ValueError("Instance users_cache must be a mapping like object.")
if "http://" in self.redirect_uri:
os.environ["OAUTHLIB_INSECURE_TRANSPORT"] = "true"
app.discord = self
@property
def user_id(self) -> typing.Union[int, None]:
"""A property which returns Discord user ID if it exists in quart :py:attr:`quart.session` object.
Returns
-------
int
The Discord user ID of current user.
None
If the user ID doesn't exists in quart :py:attr:`quart.session`.
"""
return session.get("DISCORD_USER_ID")
@staticmethod
@abc.abstractmethod
def save_authorization_token(token: dict):
raise NotImplementedError
@staticmethod
@abc.abstractmethod
def get_authorization_token() -> dict:
raise NotImplementedError
def _fetch_token(self, state):
discord = self._make_session(state=state)
return discord.fetch_token(
configs.DISCORD_TOKEN_URL,
client_secret=self.__client_secret,
authorization_response=request.url
)
def _make_session(self, token: str = None, state: str = None, scope: list = None) -> OAuth2Session:
"""A low level method used for creating OAuth2 session.
Parameters
----------
token : str, optional
The authorization token to use which was previously received from authorization code grant.
state : str, optional
The state to use for OAuth2 session.
scope : list, optional
List of valid `Discord OAuth2 Scopes
<https://discordapp.com/developers/docs/topics/oauth2#shared-resources-oauth2-scopes>`_.
Returns
-------
OAuth2Session
An instance of OAuth2Session class.
"""
return OAuth2Session(
client_id=self.client_id,
token=token or self.get_authorization_token(),
state=state or session.get("DISCORD_OAUTH2_STATE"),
scope=scope,
redirect_uri=self.redirect_uri,
auto_refresh_kwargs={
'client_id': self.client_id,
'client_secret': self.__client_secret,
},
auto_refresh_url=configs.DISCORD_TOKEN_URL,
token_updater=self.save_authorization_token)
def request(self, route: str, method="GET", data=None, oauth=True, **kwargs) -> typing.Union[dict, str]:
"""Sends HTTP request to provided route or discord endpoint.
Note
----
It automatically prefixes the API Base URL so you will just have to pass routes or URL endpoints.
Parameters
----------
route : str
Route or endpoint URL to send HTTP request to. Example: ``/users/@me``
method : str, optional
Specify the HTTP method to use to perform this request.
data : dict, optional
The optional payload the include with the request.
oauth : bool
A boolean determining if this should be Discord OAuth2 session request or any standard request.
Returns
-------
dict, str
Dictionary containing received from sent HTTP GET request if content-type is ``application/json``
otherwise returns raw text content of the response.
Raises
------
quart_discord.Unauthorized
Raises :py:class:`quart_discord.Unauthorized` if current user is not authorized.
quart_discord.RateLimited
Raises an instance of :py:class:`quart_discord.RateLimited` if application is being rate limited by Discord.
"""
route = configs.DISCORD_API_BASE_URL + route
response = self._make_session(
).request(method, route, data, **kwargs) if oauth else requests.request(method, route, data=data, **kwargs)
if response.status_code == 401:
raise exceptions.Unauthorized
if response.status_code == 429:
raise exceptions.RateLimited(response)
try:
return response.json()
except json.JSONDecodeError:
return response.text
def bot_request(self, route: str, method="GET", **kwargs) -> typing.Union[dict, str]:
"""Make HTTP request to specified endpoint with bot token as authorization headers.
Parameters
----------
route : str
Route or endpoint URL to send HTTP request to.
method : str, optional
Specify the HTTP method to use to perform this request.
Returns
-------
dict, str
Dictionary containing received from sent HTTP GET request if content-type is ``application/json``
otherwise returns raw text content of the response.
Raises
------
quart_discord.Unauthorized
Raises :py:class:`quart_discord.Unauthorized` if current user is not authorized.
quart_discord.RateLimited
Raises an instance of :py:class:`quart_discord.RateLimited` if application is being rate limited by Discord.
"""
headers = {"Authorization": f"Bot {self.__bot_token}"}
return self.request(route, method=method, oauth=False, headers=headers, **kwargs)

229
quart_discord/client.py Normal file
View File

@@ -0,0 +1,229 @@
import jwt
import typing
import discord
from . import configs, _http, models, utils, exceptions
from oauthlib.common import add_params_to_uri
from quart import request, session, redirect, current_app
class DiscordOAuth2Session(_http.DiscordOAuth2HttpClient):
"""Main client class representing hypothetical OAuth2 session with discord.
It uses Quart `session <https://pgjones.gitlab.io/quart/reference/source/quart.sessions.html#quart.sessions.Session>`_
local proxy object to save state, authorization token and keeps record of users sessions across different requests.
This class inherits :py:class:`quart_discord._http.DiscordOAuth2HttpClient` class.
Parameters
----------
app : Quart
An instance of your `quart application <https://pgjones.gitlab.io/quart/reference/source/quart.app.html#quart.app.Quart>`_.
client_id : int, optional
The client ID of discord application provided. Can be also set to quart config
with key ``DISCORD_CLIENT_ID``.
client_secret : str, optional
The client secret of discord application provided. Can be also set to quart config
with key ``DISCORD_CLIENT_SECRET``.
redirect_uri : str, optional
The default URL to use to redirect user to after authorization. Can be also set to quart config
with key ``DISCORD_REDIRECT_URI``.
bot_token : str, optional
The bot token of the application. This is required when you also need to access bot scope resources
beyond the normal resources provided by the OAuth. Can be also set to quart config with
key ``DISCORD_BOT_TOKEN``.
users_cache : cachetools.LFUCache, optional
Any dict like mapping to internally cache the authorized users. Preferably an instance of
cachetools.LFUCache or cachetools.TTLCache. If not specified, default cachetools.LFUCache is used.
Uses the default max limit for cache if ``DISCORD_USERS_CACHE_MAX_LIMIT`` isn't specified in app config.
Attributes
----------
client_id : int
The client ID of discord application provided.
redirect_uri : str
The default URL to use to redirect user to after authorization.
users_cache : cachetools.LFUCache
A dict like mapping to internally cache the authorized users. Preferably an instance of
cachetools.LFUCache or cachetools.TTLCache. If not specified, default cachetools.LFUCache is used.
Uses the default max limit for cache if ``DISCORD_USERS_CACHE_MAX_LIMIT`` isn't specified in app config.
"""
@staticmethod
def __save_state(state):
session["DISCORD_OAUTH2_STATE"] = state
@staticmethod
def __get_state():
return session.pop("DISCORD_OAUTH2_STATE", str())
def create_session(
self, scope: list = None, *, data: dict = None, prompt: bool = True,
permissions: typing.Union[discord.Permissions, int] = 0, **params
):
"""Primary method used to create OAuth2 session and redirect users for
authorization code grant.
Parameters
----------
scope : list, optional
An optional list of valid `Discord OAuth2 Scopes
<https://discordapp.com/developers/docs/topics/oauth2#shared-resources-oauth2-scopes>`_.
data : dict, optional
A mapping of your any custom data which you want to access after authorization grant. Use
`:py:meth:quart_discord.DiscordOAuth2Session.callback` to retrieve this data in your callback view.
prompt : bool, optional
Determines if the OAuth2 grant should be explicitly prompted and re-approved. Defaults to True.
Specify False for implicit grant which will skip the authorization screen and redirect to redirect URI.
permissions: typing.Union[discord.Permissions, int], optional
An optional parameter determining guild permissions of the bot while adding it to a guild using
discord OAuth2 with `bot` scope. It is same as generating so called *bot invite link* which redirects
to your callback endpoint after bot authorization flow. Defaults to 0 or no permissions.
params : kwargs, optional
Additional query parameters to append to authorization URL for customized OAuth flow.
Returns
-------
redirect
Quart redirect to discord authorization servers to complete authorization code grant process.
"""
scope = scope or request.args.get("scope", str()).split() or configs.DISCORD_OAUTH_DEFAULT_SCOPES
if not prompt and set(scope) & set(configs.DISCORD_PASSTHROUGH_SCOPES):
raise ValueError("You should use explicit OAuth grant for passthrough scopes like bot.")
state = jwt.encode(data or dict(), current_app.config["SECRET_KEY"]).decode(encoding="utf-8")
discord_session = self._make_session(scope=scope, state=state)
authorization_url, state = discord_session.authorization_url(configs.DISCORD_AUTHORIZATION_BASE_URL)
self.__save_state(state)
params = params or dict()
params["prompt"] = "consent" if prompt else "none"
if "bot" in scope:
if not isinstance(permissions, (discord.Permissions, int)):
raise ValueError(f"Passed permissions must be an int or discord.Permissions, not {type(permissions)}.")
if isinstance(permissions, discord.Permissions):
permissions = permissions.value
params["permissions"] = permissions
try:
params["disable_guild_select"] = utils.json_bool(params["disable_guild_select"])
except KeyError:
pass
authorization_url = add_params_to_uri(authorization_url, params)
return redirect(authorization_url)
@staticmethod
def save_authorization_token(token: dict):
"""A staticmethod which saves a dict containing Discord OAuth2 token and other secrets to the user's cookies.
Meaning by default, it uses client side session handling.
Override this method if you want to handle the user's session server side. If this method is overridden then,
you must also override :py:meth:`quart_discord.DiscordOAuth2Session.get_authorization_token`.
"""
session["DISCORD_OAUTH2_TOKEN"] = token
@staticmethod
def get_authorization_token() -> dict:
"""A static method which returns a dict containing Discord OAuth2 token and other secrets which was saved
previously by `:py:meth:`quart_discord.DiscordOAuth2Session.save_authorization_token` from user's cookies.
You must override this method if you are implementing server side session handling.
"""
return session.get("DISCORD_OAUTH2_TOKEN")
def callback(self):
"""A method which should be always called after completing authorization code grant process
usually in callback view.
It fetches the authorization token and saves it quart
`session <https://pgjones.gitlab.io/quart/reference/source/quart.sessions.html#quart.sessions.Session>`_ object.
"""
error = request.values.get("error")
if error:
if error == "access_denied":
raise exceptions.AccessDenied()
raise exceptions.HttpException(error)
state = self.__get_state()
token = self._fetch_token(state)
self.save_authorization_token(token)
return jwt.decode(state, current_app.config["SECRET_KEY"])
def revoke(self):
"""This method clears current discord token, state and all session data from quart
`session <https://pgjones.gitlab.io/quart/reference/source/quart.sessions.html#quart.sessions.Session>`_. Which
means user will have to go through discord authorization token grant flow again. Also tries to remove the user
from internal cache if they exist.
"""
self.users_cache.pop(self.user_id, None)
for session_key in self.SESSION_KEYS:
try:
session.pop(session_key)
except KeyError:
pass
@property
def authorized(self):
"""A boolean indicating whether current session has authorization token or not."""
return self._make_session().authorized
@staticmethod
def fetch_user() -> models.User:
"""This method returns user object from the internal cache if it exists otherwise makes an API call to do so.
Returns
-------
quart_discord.models.User
"""
return models.User.get_from_cache() or models.User.fetch_from_api()
@staticmethod
def fetch_connections() -> list:
"""This method returns list of user connection objects from internal cache if it exists otherwise
makes an API call to do so.
Returns
-------
list
List of :py:class:`quart_discord.models.UserConnection` objects.
"""
user = models.User.get_from_cache()
try:
if user.connections is not None:
return user.connections
except AttributeError:
pass
return models.UserConnection.fetch_from_api()
@staticmethod
def fetch_guilds() -> list:
"""This method returns list of guild objects from internal cache if it exists otherwise makes an API
call to do so.
Returns
-------
list
List of :py:class:`quart_discord.models.Guild` objects.
"""
user = models.User.get_from_cache()
try:
if user.guilds is not None:
return user.guilds
except AttributeError:
pass
return models.Guild.fetch_from_api()

30
quart_discord/configs.py Normal file
View File

@@ -0,0 +1,30 @@
DISCORD_API_BASE_URL = "https://discordapp.com/api"
DISCORD_AUTHORIZATION_BASE_URL = DISCORD_API_BASE_URL + "/oauth2/authorize"
DISCORD_TOKEN_URL = DISCORD_API_BASE_URL + "/oauth2/token"
DISCORD_OAUTH_ALL_SCOPES = [
"bot", "connections", "email", "identify", "guilds", "guilds.join",
"gdm.join", "messages.read", "rpc", "rpc.api", "rpc.notifications.read", "webhook.incoming",
]
DISCORD_OAUTH_DEFAULT_SCOPES = [
"identify", "email", "guilds", "guilds.join"
]
DISCORD_PASSTHROUGH_SCOPES = [
"bot", "webhook.incoming",
]
DISCORD_IMAGE_BASE_URL = "https://cdn.discordapp.com/"
DISCORD_EMBED_BASE_BASE_URL = "https://cdn.discordapp.com/"
DISCORD_IMAGE_FORMAT = "png"
DISCORD_ANIMATED_IMAGE_FORMAT = "gif"
DISCORD_USER_AVATAR_BASE_URL = DISCORD_IMAGE_BASE_URL + "avatars/{user_id}/{avatar_hash}.{format}"
DISCORD_DEFAULT_USER_AVATAR_BASE_URL = DISCORD_EMBED_BASE_BASE_URL + "embed/avatars/{modulo5}.png"
DISCORD_GUILD_ICON_BASE_URL = DISCORD_IMAGE_BASE_URL + "icons/{guild_id}/{icon_hash}.png"
DISCORD_USERS_CACHE_DEFAULT_MAX_LIMIT = 100

48
quart_discord/exceptions.py Normal file
View File

@@ -0,0 +1,48 @@
import json
class HttpException(Exception):
"""Base Exception class representing a HTTP exception."""
class RateLimited(HttpException):
"""A HTTP Exception raised when the application is being rate limited.
It provides the ``response`` attribute which can be used to get more details of the actual response from
the Discord API with few more shorthands to ``response.json()``.
Attributes
----------
response : requests.Response
The actual response object received from Discord API.
json : dict
The actual JSON data received. Shorthand to ``response.json()``.
message : str
A message saying you are being rate limited.
retry_after : int
The number of milliseconds to wait before submitting another request.
is_global : bool
A value indicating if you are being globally rate limited or not
"""
def __init__(self, response):
self.response = response
try:
self.json = self.response.json()
except json.JSONDecodeError:
self.json = dict()
self.message = self.response.text
else:
self.message = self.json["message"]
self.is_global = self.json["global"]
self.retry_after = self.json["retry_after"]
finally:
super().__init__(self.message)
class Unauthorized(HttpException):
"""A HTTP Exception raised when user is not authorized."""
class AccessDenied(HttpException):
"""Exception raised when user cancels OAuth authorization grant."""

12
quart_discord/models/__init__.py Normal file
View File

@@ -0,0 +1,12 @@
from .connections import UserConnection
from .integration import Integration
from .user import User, Bot
from .guild import Guild
__all__ = [
"Guild",
"User",
"Bot",
"UserConnection",
]

65
quart_discord/models/base.py Normal file
View File

@@ -0,0 +1,65 @@
from quart import current_app
from abc import ABCMeta, abstractmethod
class DiscordModelsMeta(ABCMeta):
ROUTE = str()
def __init__(cls, name, *args, **kwargs):
if not cls.ROUTE and name != "DiscordModelsBase":
raise NotImplementedError(f"ROUTE must be specified in a Discord model: {name}.")
super().__init__(name, *args, **kwargs)
class DiscordModelsBase(metaclass=DiscordModelsMeta):
BOT = False
MANY = False
@abstractmethod
def __init__(self, payload):
self._payload = payload
@staticmethod
def _request(*args, **kwargs):
"""A shorthand to :py:func:quart_discord.request`. It uses Quart current_app local proxy to get the
Quart-Discord client.
"""
return current_app.discord.request(*args, **kwargs)
@staticmethod
def _bot_request(*args, **kwargs):
"""A shorthand to :py:func:quart_discord.bot_request`."""
return current_app.discord.bot_request(*args, **kwargs)
@classmethod
def fetch_from_api(cls):
"""A class method which returns an instance or list of instances of this model by implicitly making an
API call to Discord.
Returns
-------
cls
An instance of this model itself.
[cls, ...]
List of instances of this model when many of these models exist.
"""
request_method = cls._bot_request if cls.BOT else cls._request
payload = request_method(cls.ROUTE)
if cls.MANY:
return [cls(_) for _ in payload]
return cls(payload)
def to_json(self):
"""A utility method which returns raw payload object as it was received from discord.
Returns
-------
dict
A dict representing raw payload object received from discord.
"""
return self._payload

86
quart_discord/models/connections.py Normal file
View File

@@ -0,0 +1,86 @@
from quart import current_app
from .base import DiscordModelsBase
from .integration import Integration
class UserConnection(DiscordModelsBase):
"""Class representing connections in discord account of the user.
Attributes
----------
id : str
ID of the connection account.
name : str
The username of the connection account.
type : str
The service of connection (twitch, youtube).
revoked : bool
A boolean representing whether the connection is revoked.
integrations : list
A list of server Integration objects.
verified : bool
A boolean representing whether the connection is verified.
friend_sync : bool
A boolean representing whether friend sync is enabled for this connection.
show_activity : bool
A boolean representing whether activities related to this connection will
be shown in presence updates.
visibility : int
An integer representing
`visibility <https://discordapp.com/developers/docs/resources/user#user-object-visibility-types>`_
of this connection.
"""
MANY = True
ROUTE = "/users/@me/connections"
def __init__(self, payload):
super().__init__(payload)
self.id = self._payload["id"]
self.name = self._payload.get("name")
self.type = self._payload.get("type")
self.revoked = self._payload.get("revoked")
self.integrations = self.__get_integrations()
self.verified = self._payload.get("verified")
self.friend_sync = self._payload.get("friend_sync")
self.show_activity = self._payload.get("show_activity")
self.visibility = self._payload.get("visibility")
def __get_integrations(self):
return [Integration(payload) for payload in self._payload.get("integrations", list())]
@property
def is_visible(self):
"""A property returning bool if this integration is visible to everyone."""
return bool(self.visibility)
@classmethod
def fetch_from_api(cls, cache=True):
"""A class method which returns an instance or list of instances of this model by implicitly making an
API call to Discord. If an instance of :py:class:`quart_discord.User` exists in the users internal cache
who are attached to these connections then, the cached property :py:attr:`quart_discord.User.connections`
is updated.
Parameters
----------
cache : bool
Determines if the :py:attr:`quart_discord.User.guilds` cache should be updated with the new guilds.
Returns
-------
list[quart_discord.UserConnection, ...]
List of instances of :py:class:`quart_discord.UserConnection` to which this user belongs.
"""
connections = super().fetch_from_api()
if cache:
user = current_app.discord.users_cache.get(current_app.discord.user_id)
try:
user.connections = connections
except AttributeError:
pass
return connections

94
quart_discord/models/guild.py Normal file
View File

@@ -0,0 +1,94 @@
from .base import DiscordModelsBase
from quart import current_app
import discord
from .. import configs
class Guild(DiscordModelsBase):
"""Class representing discord Guild the user is part of.
Operations
----------
x == y
Checks if two guild's are the same.
x != y
Checks if two guild's are not the same.
str(x)
Returns the guild's name.
Attributes
----------
id : int
Discord ID of the guild.
name : str
Name of the guild.
icon_hash : str
Hash of guild's icon.
is_owner : bool
Boolean determining if current user is owner of the guild or not.
permissions : discord.Permissions
An instance of discord.Permissions representing permissions of current user in the guild.
"""
MANY = True
ROUTE = "/users/@me/guilds"
def __init__(self, payload):
super().__init__(payload)
self.id = int(self._payload["id"])
self.name = self._payload["name"]
self.icon_hash = self._payload.get("icon")
self.is_owner = self._payload.get("owner")
self.permissions = self.__get_permissions(self._payload.get("permissions"))
@staticmethod
def __get_permissions(permissions_value):
if permissions_value is None:
return
return discord.Permissions(int(permissions_value))
def __str__(self):
return self.name
def __eq__(self, guild):
return isinstance(guild, Guild) and guild.id == self.id
def __ne__(self, guild):
return not self.__eq__(guild)
@property
def icon_url(self):
"""A property returning direct URL to the guild's icon. Returns None if guild has no icon set."""
if not self.icon_hash:
return
return configs.DISCORD_GUILD_ICON_BASE_URL.format(guild_id=self.id, icon_hash=self.icon_hash)
@classmethod
def fetch_from_api(cls, cache=True):
"""A class method which returns an instance or list of instances of this model by implicitly making an
API call to Discord. If an instance of :py:class:`quart_discord.User` exists in the users internal cache
who belongs to these guilds then, the cached property :py:attr:`quart_discord.User.guilds` is updated.
Parameters
----------
cache : bool
Determines if the :py:attr:`quart_discord.User.guilds` cache should be updated with the new guilds.
Returns
-------
list[quart_discord.Guild, ...]
List of instances of :py:class:`quart_discord.Guild` to which this user belongs.
"""
guilds = super().fetch_from_api()
if cache:
user = current_app.discord.users_cache.get(current_app.discord.user_id)
try:
user.guilds = {guild.id: guild for guild in guilds}
except AttributeError:
pass
return guilds

42
quart_discord/models/integration.py Normal file
View File

@@ -0,0 +1,42 @@
class Integration(object):
""""Class representing discord server integrations.
Attributes
----------
id : int
Integration ID.
name : str
Name of integration.
type : str
Integration type (twitch, youtube, etc).
enabled : bool
A boolean representing if this integration is enabled.
syncing : bool
A boolean representing if this integration is syncing.
role_id : int
ID that this integration uses for subscribers.
expire_behaviour : int
An integer representing the behaviour of expiring subscribers.
expire_grace_period : int
An integer representing the grace period before expiring subscribers.
account : dict
A dictionary representing raw
`account <https://discordapp.com/developers/docs/resources/guild#integration-account-object>`_ object.
synced_at : ISO8601 timestamp
Representing when this integration was last synced.
"""
def __init__(self, payload):
self._payload = payload
self.id = int(self._payload.get("id", 0))
self.name = self._payload.get("name")
self.type = self._payload.get("type")
self.enabled = self._payload.get("enabled")
self.syncing = self._payload.get("syncing")
self.role_id = int(self._payload.get("role_id", 0))
self.expire_behaviour = self._payload.get("expire_behaviour")
self.expire_grace_period = self._payload.get("expire_grace_period")
# self.user = User(self._payload.get("user", dict()))
self.account = self._payload.get("account")
self.synced_at = self._payload.get("synced_at")

226
quart_discord/models/user.py Normal file
View File

@@ -0,0 +1,226 @@
from .. import configs
from .guild import Guild
from .. import exceptions
from .base import DiscordModelsBase
from .connections import UserConnection
from quart import current_app, session
class User(DiscordModelsBase):
"""Class representing Discord User.
Operations
----------
x == y
Checks if two user's are the same.
x != y
Checks if two user's are not the same.
str(x)
Returns the user's name with discriminator.
Attributes
----------
id : int
The discord ID of the user.
username : str
The discord username of the user.
discriminator : str
4 length string representing discord tag of the user.
avatar_hash : str
Hash of users avatar.
bot : bool
A boolean representing whether the user belongs to an OAuth2 application.
mfa_enabled : bool
A boolean representing whether the user has two factor enabled on their account.
locale : str
The user's chosen language option.
verified : bool
A boolean representing whether the email on this account has been verified.
email : str
User's email ID.
flags : int
An integer representing the
`user flags <https://discordapp.com/developers/docs/resources/user#user-object-user-flags>`_.
premium_type : int
An integer representing the
`type of nitro subscription <https://discordapp.com/developers/docs/resources/user#user-object-premium-types>`_.
connections : list
A list of :py:class:`quart_discord.UserConnection` instances. These are cached and this list might be empty.
"""
ROUTE = "/users/@me"
def __init__(self, payload):
super().__init__(payload)
self.id = int(self._payload["id"])
self.username = self._payload["username"]
self.discriminator = self._payload["discriminator"]
self.avatar_hash = self._payload.get("avatar", self.discriminator)
self.bot = self._payload.get("bot", False)
self.mfa_enabled = self._payload.get("mfa_enabled")
self.locale = self._payload.get("locale")
self.verified = self._payload.get("verified")
self.email = self._payload.get("email")
self.flags = self._payload.get("flags")
self.premium_type = self._payload.get("premium_type")
# Few properties which are intended to be cached.
self._guilds = None # Mapping of guild ID to quart_discord.models.Guild(...).
self.connections = None # List of quart_discord.models.UserConnection(...).
@property
def guilds(self):
"""A cached mapping of user's guild ID to :py:class:`quart_discord.Guild`. The guilds are cached when the first
API call for guilds is requested so it might be an empty dict.
"""
try:
return list(self._guilds.values())
except AttributeError:
pass
@guilds.setter
def guilds(self, value):
self._guilds = value
def __str__(self):
return f"{self.name}#{self.discriminator}"
def __eq__(self, user):
return isinstance(user, User) and user.id == self.id
def __ne__(self, user):
return not self.__eq__(user)
@property
def name(self):
"""An alias to the username attribute."""
return self.username
@property
def avatar_url(self):
"""A property returning direct URL to user's avatar."""
if not self.avatar_hash:
return
image_format = configs.DISCORD_ANIMATED_IMAGE_FORMAT \
if self.is_avatar_animated else configs.DISCORD_IMAGE_FORMAT
return configs.DISCORD_USER_AVATAR_BASE_URL.format(
user_id=self.id, avatar_hash=self.avatar_hash, format=image_format)
@property
def default_avatar_url(self):
"""A property which returns the default avatar URL as when user doesn't has any avatar set."""
return configs.DISCORD_DEFAULT_USER_AVATAR_BASE_URL.format(modulo5=int(self.discriminator) % 5)
@property
def is_avatar_animated(self):
"""A boolean representing if avatar of user is animated. Meaning user has GIF avatar."""
try:
return self.avatar_hash.startswith("a_")
except AttributeError:
return False
@classmethod
def fetch_from_api(cls, guilds=False, connections=False):
"""A class method which returns an instance of this model by implicitly making an
API call to Discord. The user returned from API will always be cached and update in internal cache.
Parameters
----------
guilds : bool
A boolean indicating if user's guilds should be cached or not. Defaults to ``False``. If chose to not
cache, user's guilds can always be obtained from :py:func:`quart_discord.Guilds.fetch_from_api()`.
connections : bool
A boolean indicating if user's connections should be cached or not. Defaults to ``False``. If chose to not
cache, user's connections can always be obtained from :py:func:`quart_discord.Connections.fetch_from_api()`.
Returns
-------
cls
An instance of this model itself.
[cls, ...]
List of instances of this model when many of these models exist."""
self = super().fetch_from_api()
current_app.discord.users_cache.update({self.id: self})
session["DISCORD_USER_ID"] = self.id
if guilds:
self.fetch_guilds()
if connections:
self.fetch_connections()
return self
@classmethod
def get_from_cache(cls):
"""A class method which returns an instance of this model if it exists in internal cache.
Returns
-------
quart_discord.User
An user instance if it exists in internal cache.
None
If the current doesn't exists in internal cache.
"""
return current_app.discord.users_cache.get(current_app.discord.user_id)
def add_to_guild(self, guild_id) -> dict:
"""Method to add user to the guild, provided OAuth2 session has already been created with ``guilds.join`` scope.
Parameters
----------
guild_id : int
The ID of the guild you want this user to be added.
Returns
-------
dict
A dict of guild member object. Returns an empty dict if user is already present in the guild.
Raises
------
quart_discord.Unauthorized
Raises :py:class:`quart_discord.Unauthorized` if current user is not authorized.
"""
try:
data = {"access_token": current_app.discord.get_authorization_token()["access_token"]}
except KeyError:
raise exceptions.Unauthorized
return self._bot_request(f"/guilds/{guild_id}/members/{self.id}", method="PUT", json=data) or dict()
def fetch_guilds(self) -> list:
"""A method which makes an API call to Discord to get user's guilds. It prepares the internal guilds cache
and returns list of all guilds the user is member of.
Returns
-------
list
List of :py:class:`quart_discord.Guilds` instances.
"""
self._guilds = {guild.id: guild for guild in Guild.fetch_from_api(cache=False)}
return self.guilds
def fetch_connections(self) -> list:
"""A method which makes an API call to Discord to get user's connections. It prepares the internal connection
cache and returns list of all connection instances.
Returns
-------
list
A list of :py:class:`quart_discord.UserConnection` instances.
"""
self.connections = UserConnection.fetch_from_api(cache=False)
return self.connections
class Bot(User):
"""Class representing the client user itself."""
# TODO: What is this?

51
quart_discord/utils.py Normal file
View File

@@ -0,0 +1,51 @@
"""Few utility functions and decorators."""
import functools
from . import exceptions
from quart import current_app
class JSONBool(object):
def __init__(self, value):
self.value = bool(value)
def __bool__(self):
return self.value
def __str__(self):
return "true" if self else "false"
@classmethod
def from_string(cls, value):
if value.lower() == "true":
return cls(True)
if value.lower() == "false":
return cls(False)
raise ValueError
def json_bool(value):
if isinstance(value, str):
return str(JSONBool.from_string(value))
return str(JSONBool(value))
# Decorators.
def requires_authorization(view):
"""A decorator for quart views which raises exception :py:class:`quart_discord.Unauthorized` if the user
is not authorized from Discord OAuth2.
"""
# TODO: Add support to validate scopes.
@functools.wraps(view)
def wrapper(*args, **kwargs):
if not current_app.discord.authorized:
raise exceptions.Unauthorized
return view(*args, **kwargs)
return wrapper