Проверка существование юзера, логин, логаут (#3)

.github/workflows/build_and_publish.yml

@@ -0,0 +1,30 @@
+name: Publish Python 🐍 distributions 📦
+
+on:
+  push:
+    branches: ['main']
+    tags:
+      - 'v*'
+
+jobs:
+  build-and-publish:
+    name: Build and publish Python 🐍 distributions 📦 to PyPI and TestPyPI
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@master
+      - name: Set up Python 3.10
+        uses: actions/setup-python@v3
+        with:
+          python-version: "3.10"
+      - name: Install dependencies
+          run: |
+            python -m ensurepip \
+            pip install build --user \
+            python -m build --wheel --sdist --outdir dist/
+      - name: Publish distribution 📦 to PyPI
+          if: startsWith(github.ref, 'refs/tags')
+          uses: pypa/gh-action-pypi-publish@release/v1
+          with:
+            repository_url: https://github.com/profcomff/auth-lib
+            username: ${{ github.actor }}
+            password: ${{ secrets.GITHUB_TOKEN }}

auth_lib/aiomethods.py

@@ -0,0 +1,53 @@
+from typing import Any
+
+import aiohttp
+
+from .exceptions import SessionExpired, AuthFailed, IncorrectData, NotFound
+
+
+class AsyncAuthLib:
+    url: str
+
+    def __init__(self, url: str):
+        self.url = url
+
+    async def email_login(self, email: str, password: str) -> dict[str, Any]:
+        json = {
+            "email": email,
+            "password": password
+        }
+        async with aiohttp.ClientSession() as session:
+            response = await session.post(url=f"{self.url}/email/login", json=json)
+        match response.status:
+            case 200:
+                return await response.json()
+            case 401:
+                raise AuthFailed(response=await response.json())
+
+    async def check_token(self, token: str) -> dict[str, Any]:
+        headers = {"token": token}
+        fields = frozenset(["email"])
+        async with aiohttp.ClientSession() as session:
+            response = await session.post(url=f"{self.url}/me", headers=headers)
+        match response.status:
+            case 200:
+                return await response.json()
+            case 400:
+                raise IncorrectData(response=await response.json())
+            case 404:
+                raise NotFound(response=await response.json())
+            case 403:
+                raise SessionExpired(response=await response.json())
+
+    async def logout(self, token: str) -> bool:
+        headers = {"token": token}
+        async with aiohttp.ClientSession() as session:
+            response = await session.post(url=f"{self.url}/logout", headers=headers)
+
+        match response.status:
+            case 200:
+                return True
+            case 401:
+                raise AuthFailed(response=await response.json())
+            case 403:
+                raise SessionExpired(response=await response.json())

auth_lib/exceptions.py

@@ -0,0 +1,23 @@
+class AuthError(Exception):
+    def __init__(self, *args, **kwargs):
+        super().__init__("Auth failed")
+
+
+class IncorrectData(AuthError):
+    def __init__(self, response: dict):
+        super().__init__(str(response))
+
+
+class AuthFailed(AuthError):
+    def __init__(self, response: dict):
+        super().__init__(str(response))
+
+
+class SessionExpired(AuthError):
+    def __init__(self, response: dict):
+        super().__init__(str(response))
+
+
+class NotFound(AuthError):
+    def __init__(self, response: dict):
+        super().__init__(str(response))

auth_lib/fastapi.py

@@ -0,0 +1,47 @@
+from urllib.parse import urljoin
+
+import aiohttp
+from fastapi.exceptions import HTTPException
+from fastapi.openapi.models import APIKey, APIKeyIn
+from fastapi.security.base import SecurityBase
+from starlette.requests import Request
+from starlette.status import HTTP_403_FORBIDDEN
+
+
+class UnionAuth(SecurityBase):
+    model = APIKey.construct(in_=APIKeyIn.header, name="Authorization")
+    scheme_name = 'token'
+    auth_url: str
+
+    def __init__(self, auth_url: str, auto_error=True) -> None:
+        super().__init__()
+        self.auto_error = auto_error
+        self.auth_url = auth_url
+
+    def _except(self):
+        if self.auto_error:
+            raise HTTPException(
+                status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+            )
+        else:
+            return {}
+
+    async def __call__(
+            self, request: Request,
+    ) -> dict[str, str]:
+        token = request.headers.get("token")
+        if not token:
+            return self._except()
+        async with aiohttp.request(
+                'POST',
+                urljoin(self.auth_url, '/me'),
+                headers={"token": token},
+        ) as r:
+            status_code = r.status
+            user = await r.json()
+        if status_code != 200:
+            self._except()
+        return user
+
+
+

auth_lib/methods.py

@@ -0,0 +1,53 @@
+from typing import Any
+
+import requests
+
+from .exceptions import SessionExpired, AuthFailed, IncorrectData, NotFound
+
+
+# See docs on https://auth.api.test.profcomff.com/docs
+
+
+class AuthLib:
+    url: str
+
+    def __init__(self, url: str):
+        self.url = url
+
+    def email_login(self, email: str, password: str) -> dict[str, Any]:
+        json = {
+            "email": email,
+            "password": password
+        }
+        response = requests.post(url=f"{self.url}/email/login", json=json)
+        match response.status_code:
+            case 200:
+                return response.json()
+            case 401:
+                raise AuthFailed(response=response.json()["body"])
+
+    def check_token(self, token: str) -> dict[str, Any]:
+        headers = {"token": token}
+        fields = frozenset(["email"])
+        response = requests.post(url=f"{self.url}/me", headers=headers)
+        match response.status_code:
+            case 200:
+                return response.json()
+            case 400:
+                raise IncorrectData(response=response.json()["body"])
+            case 404:
+                raise NotFound(response=response.json()["body"])
+            case 403:
+                raise SessionExpired(response=response.json()["body"])
+
+    def logout(self, token: str) -> bool:
+        headers = {"token": token}
+        response = requests.post(url=f"{self.url}/logout", headers=headers)
+
+        match response.status_code:
+            case 200:
+                return True
+            case 401:
+                raise AuthFailed(response=response.json()["body"])
+            case 403:
+                raise SessionExpired(response=response.json()["body"])

requirements.dev.txt

@@ -0,0 +1,5 @@
+fastapi
+starlette
+requests
+aiohttp
+setuptools

setup.py

@@ -0,0 +1,21 @@
+from setuptools import setup, find_packages
+
+with open("README.md", "r") as readme_file:
+    readme = readme_file.read()
+
+setup(
+    name="auth_lib",
+    version="0.0.1",
+    author="Semyon Grigoriev",
+    long_description=readme,
+    long_description_content_type="text/markdown",
+    url="https://github.com/profcomff/auth-lib",
+    packages=find_packages(),
+    install_requires=["requests", "aiohttp", "setuptools"],
+    extras_require={
+        "fastapi": ["fastapi", "starlette"],
+    },
+    classifiers=[
+        "Programming Language :: Python :: 3.11",
+    ],
+)