feat: AuthClientクラスを追加 - issue #7

CHANGELOG.md

@@ -0,0 +1,10 @@
+# Change Log
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](http://keepachangelog.com/)
+and this project adheres to [Semantic Versioning](http://semver.org/).
+
+## [Unreleased]
+### Added
+
+- utils.pyに `AuthClient` クラスを追加しました

mipac/util.py

@@ -2,11 +2,16 @@
 MiPAを使用する上でちょっとした際に便利なツール一覧
 """
 from __future__ import annotations
+import asyncio
 
 import re
 from datetime import datetime, timedelta
 from inspect import isawaitable
 from typing import Any, Callable, Dict, Iterable, List, Optional, TypeVar
+from urllib.parse import urlencode
+import uuid
+
+import aiohttp
 
 __all__ = (
     'deprecated_func',
@@ -36,6 +41,125 @@ class MiTime:
         self.end = end
 
 
+class AuthClient:
+    """
+    Tokenの取得を手助けするクラス
+    """
+    
+    def __init__(
+        self,
+        instance_uri: str,
+        name: str,
+        description: str,
+        permissions: Optional[List[str]] = None,
+        *,
+        icon: Optional[str] = None,
+        use_miauth: bool = False,
+    ):
+        """
+        Parameters
+        ----------
+        instance_uri : str
+            アプリケーションを作成したいインスタンスのURL
+        name : str
+            アプリケーションの名前
+        description : str
+            アプリケーションの説明
+        permissions : Optional[List[str]], default=None
+            アプリケーションが要求する権限
+        icon: Optional[str], default=None
+            アプリケーションのアイコン画像URL
+        use_miauth: bool, default=False
+            MiAuthを使用するか
+        """
+        
+        if permissions is None:
+            permissions = ['read:account']
+        self.__client_session = aiohttp.ClientSession()
+        self.__instance_uri: str = instance_uri
+        self.__name: str = name
+        self.__description: str = description
+        self.__permissions: List[str] = permissions
+        self.__icon: Optional[str] = icon
+        self.__use_miauth: bool = use_miauth
+        self.__session_token: uuid.UUID
+        self.__secret: str
+
+    async def get_auth_url(self) -> str:
+        """
+        認証に使用するURLを取得します
+        
+        Returns
+        -------
+        str
+            認証に使用するURL
+        """
+        
+        field = remove_dict_empty(
+            {
+                'name': self.__name,
+                'description': self.__description,
+                'icon': self.__icon,
+            }
+        )
+        if self.__use_miauth:
+            field['permissions'] = self.__permissions
+            query = urlencode(field)
+            self.__session_token = uuid.uuid4()
+            return (
+                f'{self.__instance_uri}/miauth/{self.__session_token}?{query}'
+            )
+        else:
+            field['permission'] = self.__permissions
+            async with self.__client_session.post(
+                f'{self.__instance_uri}/api/app/create', json=field
+            ) as res:
+                data = await res.json()
+                self.__secret = data['secret']
+            async with self.__client_session.post(
+                f'{self.__instance_uri}/api/auth/session/generate',
+                json={'appSecret': self.__secret},
+            ) as res:
+                data = await res.json()
+                self.__session_token = data['token']
+                return data['url']
+
+    async def check_auth(self) -> str:
+        """
+        認証が完了したかを確認し完了している場合はTokenを返します
+        
+        Returns
+        -------
+        str
+            Token
+        """
+        
+        if self.__use_miauth:
+            while True:
+                async with self.__client_session.post(
+                    f'{self.__instance_uri}/api/miauth/{self.__session_token}/check'
+                ) as res:
+                    data = await res.json()
+                    if data.get('ok') is True:
+                        break
+                await asyncio.sleep(1)
+        else:
+            while True:
+                async with self.__client_session.post(
+                    f'{self.__instance_uri}/api/auth/session/userkey',
+                    json={
+                        'appSecret': self.__secret,
+                        'token': self.__session_token,
+                    },
+                ) as res:
+                    data = await res.json()
+                    if data.get('error', {}).get('code') != 'PENDING_SESSION':
+                        break
+                await asyncio.sleep(1)
+        await self.__client_session.close()
+        return data['token'] if self.__use_miauth else data['accessToken']
+
+
 def get_cache_key(func):
     async def decorator(self, *args, **kwargs):
         ordered_kwargs = sorted(kwargs.items())