fix: mypy errors on custom models (#66)

* fix: mypy errors on custom models

see the issue #63

while inheriting from models (Token, Client, AuthorizationCode), and using them in aioauth, mypy throws an error like:

```
error: Argument 1 of "create_authorization_code" is incompatible with supertype "BaseStorage"; supertype defines the argument type as "Request"
```

this PR fixes the above bug by adding additional `TToken`, `TClient` and `TAuthorizationCode` parameters to the `BaseModel` generic.

usage example:

```python
from dataclasses import dataclass
from aioauth_fastapi.router import get_oauth2_router
from aioauth.storage import BaseStorage
from aioauth.requests import BaseRequest
from aioauth.models import AuthorizationCode, Client, Token
from aioauth.config import Settings
from aioauth.server import AuthorizationServer
from fastapi import FastAPI

app = FastAPI()

@DataClass
class User:
    """Custom user model"""
    first_name: str
    last_name: str


@DataClass
class Request(BaseRequest[Query, Post, User]):
    """"Custom request"""


class Storage(BaseStorage[Token, Client, AuthorizationCode, Request]):
    """
    Storage methods must be implemented here.
    """

storage = Storage()
authorization_server = AuthorizationServer[Request, Storage](storage)

# NOTE: Redefinition of the default aioauth settings
# INSECURE_TRANSPORT must be enabled for local development only!
settings = Settings(
    INSECURE_TRANSPORT=True,
)

# Include FastAPI router with oauth2 endpoints.
app.include_router(
    get_oauth2_router(authorization_server, settings),
    prefix="/oauth2",
    tags=["oauth2"],
)
```

* enums were replaced to literals

Author: aliev

aioauth/collections.py

@@ -33,8 +33,8 @@ class HTTPHeaderDict(UserDict):
         d['hElLo'] == 'world' # >>> True
     """
 
-    def __setitem__(self, key, value):
+    def __setitem__(self, key: str, value: str):
         super().__setitem__(key.lower(), value)
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: str):
         return super().__getitem__(key.lower())

aioauth/errors.py

@@ -9,16 +9,17 @@
 """
 
 from http import HTTPStatus
-from typing import Optional
+from typing import Generic, Optional
 from urllib.parse import urljoin
+from typing_extensions import Literal
 
 from .collections import HTTPHeaderDict
 from .constances import default_headers
-from .requests import Request
+from .requests import TRequest
 from .types import ErrorType
 
 
-class OAuth2Error(Exception):
+class OAuth2Error(Exception, Generic[TRequest]):
     """Base exception that all other exceptions inherit from."""
 
     error: ErrorType
@@ -29,7 +30,7 @@ class OAuth2Error(Exception):
 
     def __init__(
         self,
-        request: Request,
+        request: TRequest,
         description: Optional[str] = None,
         headers: Optional[HTTPHeaderDict] = None,
     ):
@@ -47,28 +48,28 @@ def __init__(
         super().__init__(f"({self.error}) {self.description}")
 
 
-class MethodNotAllowedError(OAuth2Error):
+class MethodNotAllowedError(OAuth2Error[TRequest]):
     """
     The request is valid, but the method trying to be accessed is not
     available to the resource owner.
     """
 
     description = "HTTP method is not allowed."
     status_code: HTTPStatus = HTTPStatus.METHOD_NOT_ALLOWED
-    error = ErrorType.METHOD_IS_NOT_ALLOWED
+    error: Literal["method_is_not_allowed"] = "method_is_not_allowed"
 
 
-class InvalidRequestError(OAuth2Error):
+class InvalidRequestError(OAuth2Error[TRequest]):
     """
     The request is missing a required parameter, includes an invalid
     parameter value, includes a parameter more than once, or is
     otherwise malformed.
     """
 
-    error = ErrorType.INVALID_REQUEST
+    error: Literal["invalid_request"] = "invalid_request"
 
 
-class InvalidClientError(OAuth2Error):
+class InvalidClientError(OAuth2Error[TRequest]):
     """
     Client authentication failed (e.g. unknown client, no client
     authentication included, or unsupported authentication method).
@@ -81,36 +82,36 @@ class InvalidClientError(OAuth2Error):
     client.
     """
 
-    error = ErrorType.INVALID_CLIENT
+    error: Literal["invalid_client"] = "invalid_client"
     status_code: HTTPStatus = HTTPStatus.UNAUTHORIZED
 
 
-class InsecureTransportError(OAuth2Error):
+class InsecureTransportError(OAuth2Error[TRequest]):
     """An exception will be thrown if the current request is not secure."""
 
     description = "OAuth 2 MUST utilize https."
-    error = ErrorType.INSECURE_TRANSPORT
+    error: Literal["insecure_transport"] = "insecure_transport"
 
 
-class UnsupportedGrantTypeError(OAuth2Error):
+class UnsupportedGrantTypeError(OAuth2Error[TRequest]):
     """
     The authorization grant type is not supported by the authorization
     server.
     """
 
-    error = ErrorType.UNSUPPORTED_GRANT_TYPE
+    error: Literal["unsupported_grant_type"] = "unsupported_grant_type"
 
 
-class UnsupportedResponseTypeError(OAuth2Error):
+class UnsupportedResponseTypeError(OAuth2Error[TRequest]):
     """
     The authorization server does not support obtaining an authorization
     code using this method.
     """
 
-    error = ErrorType.UNSUPPORTED_RESPONSE_TYPE
+    error: Literal["unsupported_response_type"] = "unsupported_response_type"
 
 
-class InvalidGrantError(OAuth2Error):
+class InvalidGrantError(OAuth2Error[TRequest]):
     """
     The provided authorization grant (e.g. authorization code, resource
     owner credentials) or refresh token is invalid, expired, revoked, does
@@ -120,53 +121,53 @@ class InvalidGrantError(OAuth2Error):
     See `RFC6749 section 5.2 <https://tools.ietf.org/html/rfc6749#section-5.2>`_.
     """
 
-    error = ErrorType.INVALID_GRANT
+    error: Literal["invalid_grant"] = "invalid_grant"
 
 
-class MismatchingStateError(OAuth2Error):
+class MismatchingStateError(OAuth2Error[TRequest]):
     """Unable to securely verify the integrity of the request and response."""
 
     description = "CSRF Warning! State not equal in request and response."
-    error = ErrorType.MISMATCHING_STATE
+    error: Literal["mismatching_state"] = "mismatching_state"
 
 
-class UnauthorizedClientError(OAuth2Error):
+class UnauthorizedClientError(OAuth2Error[TRequest]):
     """
     The authenticated client is not authorized to use this authorization
     grant type.
     """
 
-    error = ErrorType.UNAUTHORIZED_CLIENT
+    error: Literal["unauthorized_client"] = "unauthorized_client"
 
 
-class InvalidScopeError(OAuth2Error):
+class InvalidScopeError(OAuth2Error[TRequest]):
     """
     The requested scope is invalid, unknown, or malformed, or
     exceeds the scope granted by the resource owner.
 
     See `RFC6749 section 5.2 <https://tools.ietf.org/html/rfc6749#section-5.2>`_.
     """
 
-    error = ErrorType.INVALID_SCOPE
+    error: Literal["invalid_scope"] = "invalid_scope"
 
 
-class ServerError(OAuth2Error):
+class ServerError(OAuth2Error[TRequest]):
     """
     The authorization server encountered an unexpected condition that
     prevented it from fulfilling the request.  (This error code is needed
     because a ``HTTP 500`` (Internal Server Error) status code cannot be returned
     to the client via a HTTP redirect.)
     """
 
-    error = ErrorType.SERVER_ERROR
+    error: Literal["temporarily_unavailable"] = "temporarily_unavailable"
 
 
-class TemporarilyUnavailableError(OAuth2Error):
+class TemporarilyUnavailableError(OAuth2Error[TRequest]):
     """
     The authorization server is currently unable to handle the request
     due to a temporary overloading or maintenance of the server.
     (This error code is needed because a ``HTTP 503`` (Service Unavailable)
     status code cannot be returned to the client via a HTTP redirect.)
     """
 
-    error = ErrorType.TEMPORARILY_UNAVAILABLE
+    error: Literal["temporarily_unavailable"] = "temporarily_unavailable"

aioauth/grant_type.py

@@ -7,6 +7,7 @@
 
 ----
 """
+from typing import Generic
 from .errors import (
     InvalidGrantError,
     InvalidRequestError,
@@ -15,22 +16,22 @@
     UnauthorizedClientError,
 )
 from .models import Client
-from .requests import Request
+from .requests import TRequest
 from .responses import TokenResponse
-from .storage import BaseStorage
+from .storage import TStorage
 from .utils import enforce_list, enforce_str, generate_token
 
 
-class GrantTypeBase:
+class GrantTypeBase(Generic[TRequest, TStorage]):
     """Base grant type that all other grant types inherit from."""
 
-    def __init__(self, storage: BaseStorage, client_id: str, client_secret: str):
+    def __init__(self, storage: TStorage, client_id: str, client_secret: str):
         self.storage = storage
         self.client_id = client_id
         self.client_secret = client_secret
 
     async def create_token_response(
-        self, request: Request, client: Client
+        self, request: TRequest, client: Client
     ) -> TokenResponse:
         """Creates token response to reply to client."""
         token = await self.storage.create_token(
@@ -50,27 +51,27 @@ async def create_token_response(
             token_type=token.token_type,
         )
 
-    async def validate_request(self, request: Request) -> Client:
+    async def validate_request(self, request: TRequest) -> Client:
         """Validates the client request to ensure it is valid."""
         client = await self.storage.get_client(
             request, client_id=self.client_id, client_secret=self.client_secret
         )
 
         if not client:
-            raise InvalidRequestError(
+            raise InvalidRequestError[TRequest](
                 request=request, description="Invalid client_id parameter value."
             )
 
         if not client.check_grant_type(request.post.grant_type):
-            raise UnauthorizedClientError(request=request)
+            raise UnauthorizedClientError[TRequest](request=request)
 
         if not client.check_scope(request.post.scope):
-            raise InvalidScopeError(request=request)
+            raise InvalidScopeError[TRequest](request=request)
 
         return client
 
 
-class AuthorizationCodeGrantType(GrantTypeBase):
+class AuthorizationCodeGrantType(GrantTypeBase[TRequest, TStorage]):
     """
     The Authorization Code grant type is used by confidential and public
     clients to exchange an authorization code for an access token. After
@@ -86,21 +87,21 @@ class AuthorizationCodeGrantType(GrantTypeBase):
         See `RFC 6749 section 1.3.1 <https://tools.ietf.org/html/rfc6749#section-1.3.1>`_.
     """
 
-    async def validate_request(self, request: Request) -> Client:
+    async def validate_request(self, request: TRequest) -> Client:
         client = await super().validate_request(request)
 
         if not request.post.redirect_uri:
-            raise InvalidRequestError(
+            raise InvalidRequestError[TRequest](
                 request=request, description="Mismatching redirect URI."
             )
 
         if not client.check_redirect_uri(request.post.redirect_uri):
-            raise InvalidRequestError(
+            raise InvalidRequestError[TRequest](
                 request=request, description="Invalid redirect URI."
             )
 
         if not request.post.code:
-            raise InvalidRequestError(
+            raise InvalidRequestError[TRequest](
                 request=request, description="Missing code parameter."
             )
 
@@ -109,30 +110,30 @@ async def validate_request(self, request: Request) -> Client:
         )
 
         if not authorization_code:
-            raise InvalidGrantError(request=request)
+            raise InvalidGrantError[TRequest](request=request)
 
         if (
             authorization_code.code_challenge
             and authorization_code.code_challenge_method
         ):
             if not request.post.code_verifier:
-                raise InvalidRequestError(
+                raise InvalidRequestError[TRequest](
                     request=request, description="Code verifier required."
                 )
 
             is_valid_code_challenge = authorization_code.check_code_challenge(
                 request.post.code_verifier
             )
             if not is_valid_code_challenge:
-                raise MismatchingStateError(request=request)
+                raise MismatchingStateError[TRequest](request=request)
 
         if authorization_code.is_expired:
-            raise InvalidGrantError(request=request)
+            raise InvalidGrantError[TRequest](request=request)
 
         return client
 
     async def create_token_response(
-        self, request: Request, client: Client
+        self, request: TRequest, client: Client
     ) -> TokenResponse:
         token_response = await super().create_token_response(request, client)
 
@@ -145,7 +146,7 @@ async def create_token_response(
         return token_response
 
 
-class PasswordGrantType(GrantTypeBase):
+class PasswordGrantType(GrantTypeBase[TRequest, TStorage]):
     """
     The Password grant type is a way to exchange a user's credentials
     for an access token. Because the client application has to collect
@@ -156,25 +157,25 @@ class PasswordGrantType(GrantTypeBase):
     disallows the password grant entirely.
     """
 
-    async def validate_request(self, request: Request) -> Client:
+    async def validate_request(self, request: TRequest) -> Client:
         client = await super().validate_request(request)
 
         if not request.post.username or not request.post.password:
-            raise InvalidRequestError(
+            raise InvalidRequestError[TRequest](
                 request=request, description="Invalid credentials given."
             )
 
         user = await self.storage.authenticate(request)
 
         if not user:
-            raise InvalidRequestError(
+            raise InvalidRequestError[TRequest](
                 request=request, description="Invalid credentials given."
             )
 
         return client
 
 
-class RefreshTokenGrantType(GrantTypeBase):
+class RefreshTokenGrantType(GrantTypeBase[TRequest, TStorage]):
     """
     The Refresh Token grant type is used by clients to exchange a
     refresh token for an access token when the access token has expired.
@@ -184,7 +185,7 @@ class RefreshTokenGrantType(GrantTypeBase):
     """
 
     async def create_token_response(
-        self, request: Request, client: Client
+        self, request: TRequest, client: Client
     ) -> TokenResponse:
         """Validate token request and create token response."""
         old_token = await self.storage.get_token(
@@ -194,7 +195,7 @@ async def create_token_response(
         )
 
         if not old_token or old_token.revoked or old_token.refresh_token_expired:
-            raise InvalidGrantError(request=request)
+            raise InvalidGrantError[TRequest](request=request)
 
         # Revoke old token
         await self.storage.revoke_token(
@@ -226,18 +227,18 @@ async def create_token_response(
             token_type=token.token_type,
         )
 
-    async def validate_request(self, request: Request) -> Client:
+    async def validate_request(self, request: TRequest) -> Client:
         client = await super().validate_request(request)
 
         if not request.post.refresh_token:
-            raise InvalidRequestError(
+            raise InvalidRequestError[TRequest](
                 request=request, description="Missing refresh token parameter."
             )
 
         return client
 
 
-class ClientCredentialsGrantType(GrantTypeBase):
+class ClientCredentialsGrantType(GrantTypeBase[TRequest, TStorage]):
     """
     The Client Credentials grant type is used by clients to obtain an
     access token outside of the context of a user. This is typically