royalnet/royalpack/tables/osu.py

from typing import *
import aiohttp
import datetime
from sqlalchemy import *
from sqlalchemy.orm import relationship, backref
from sqlalchemy.ext.declarative import declared_attr

from ..types import Updatable, oauth_refresh


# noinspection PyAttributeOutsideInit
class Osu(Updatable):
    __tablename__ = "osu"

    @declared_attr
    def user_id(self):
        return Column(Integer, ForeignKey("users.uid"))

    @declared_attr
    def user(self):
        return relationship("User", backref=backref("osu"))

    @declared_attr
    def access_token(self):
        return Column(String, nullable=False)

    @declared_attr
    def refresh_token(self):
        return Column(String, nullable=False)

    @declared_attr
    def expiration_date(self):
        return Column(DateTime, nullable=False)

    @declared_attr
    def osu_id(self):
        return Column(Integer, primary_key=True)

    @declared_attr
    def username(self):
        return Column(String, nullable=False)

    @declared_attr
    def avatar_url(self):
        return Column(String)

    @declared_attr
    def standard_pp(self):
        return Column(Float)

    @declared_attr
    def taiko_pp(self):
        return Column(Float)

    @declared_attr
    def catch_pp(self):
        return Column(Float)

    @declared_attr
    def mania_pp(self):
        return Column(Float)

    async def refresh(self, *, client_id, client_secret):
        j = await oauth_refresh(url="https://osu.ppy.sh/oauth/token",
                                client_id=client_id,
                                client_secret=client_secret,
                                refresh_code=self.refresh_token)
        self.access_token = j["access_token"]
        self.refresh_token = j.get("refresh_token") or self.refresh_token
        self.expiration_date = datetime.datetime.now() + datetime.timedelta(seconds=j["expires_in"])

    async def refresh_if_expired(self, *, client_id, client_secret):
        if datetime.datetime.now() >= self.expiration_date:
            await self.refresh(client_id=client_id, client_secret=client_secret)

    def json(self) -> dict:
        return {
            "osu_id": self.osu_id,
            "username": self.username,
            "avatar_url": self.avatar_url,
            "standard": {
                "pp": self.standard_pp,
            },
            "taiko": {
                "pp": self.taiko_pp,
            },
            "catch": {
                "pp": self.catch_pp,
            },
            "mania": {
                "pp": self.mania_pp,
            },
        }