crappy

auth/telegram.py

@@ -0,0 +1,28 @@
+import hmac
+import hashlib
+import os
+
+import settings
+
+
+def _generate_hash(telegram_data: dict):
+    data = telegram_data.copy()
+    del data['hash']
+    keys = sorted(data.keys())
+    string_arr = []
+    for key in keys:
+        if data[key] is not None:
+            string_arr.append(key + '=' + str(data[key]))
+    string_cat = '\n'.join(string_arr)
+
+    secret_key = hashlib.sha256(settings.TELEGRAM_BOT_TOKEN.encode('utf-8')).digest()
+    hash_bytes = bytes(string_cat, 'utf-8')
+    hmac_hash = hmac.new(secret_key, hash_bytes, hashlib.sha256).hexdigest()
+    return hmac_hash
+
+
+def telegram_authorize(telegram_data: dict):
+    generated_hash = _generate_hash(telegram_data)
+    user_hash = telegram_data['hash']
+    return generated_hash == user_hash
+

database/__init__.py

@@ -0,0 +1 @@
+from .models.basic import *

database/base.py

@@ -1,3 +1,6 @@
+from typing import Annotated
+
+from fastapi import Depends
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 from sqlalchemy.orm import sessionmaker, declarative_base
 
@@ -16,3 +19,6 @@ BaseModel = declarative_base()
 async def get_session() -> AsyncSession:
     async with session_maker() as session:
         yield session
+
+
+DatabaseDependency = Annotated[AsyncSession, Depends(get_session)]

database/models/basic.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Column, Integer, String
+from sqlalchemy import Column, Integer, String, BigInteger, Boolean
 
 from ..base import BaseModel
 
@@ -6,5 +6,7 @@ from ..base import BaseModel
 class User(BaseModel):
     __tablename__ = 'users'
     id = Column(Integer, autoincrement=True, primary_key=True, index=True)
-    login = Column(String, unique=True)
+    telegram_id = Column(BigInteger, nullable=False, index=True)
-    password = Column(String, unique=True)
+    phone_number = Column(String)
+
+    is_admin = Column(Boolean, nullable=False, default=False)

main.py

@@ -1,11 +1,26 @@
-from fastapi import FastAPI
+from typing import Annotated
+
+from fastapi import FastAPI, Depends
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from database.base import get_session
+from database.models import User
+import routers
 
 app = FastAPI()
+routers_list = [
+    routers.auth_router
+]
+for router in routers_list:
+    app.include_router(router)
 
 
 @app.get("/")
-async def root():
+async def root(db_session: Annotated[AsyncSession, Depends(get_session)]):
-    return {"message": "Hello World"}
+    user: User = await db_session.scalar(select(User).where(User.id == 1))
+
+    return {"message": user.login}
 
 
 @app.get("/hello/{name}")

migrate.bat

@@ -0,0 +1,3 @@
+alembic.exe revision --autogenerate
+alembic.exe upgrade head
+

migrate.sh

@@ -0,0 +1,3 @@
+alembic.exe revision --autogenerate
+alembic.exe upgrade head
+

requirements.txt

@@ -1,6 +1,7 @@
 # FastApi
 fastapi
 pydantic
+uvicorn==0.20.0
 
 # Security
 python-jose[cryptography]

routers/__init__.py

@@ -0,0 +1 @@
+from .auth import auth_router

routers/auth.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter
+from sqlalchemy import select, insert
+
+import database
+from database import User
+from database.base import DatabaseDependency
+from schemas.auth.requests import *
+from auth.telegram import telegram_authorize
+from schemas.auth.responses import AuthLoginResponse
+
+auth_router = APIRouter(
+    prefix='/auth',
+    tags=['auth'],
+)
+
+
+@auth_router.post('/login', response_model=AuthLoginResponse)
+async def login(request: AuthLoginRequest, db_session: DatabaseDependency):
+    existing_user: User = await db_session.scalar(select(User).where(User.telegram_id == request.id))
+    return AuthLoginResponse(ok=True, jwt_token="dasdasd")

schemas/auth/requests.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel
+
+
+class AuthLoginRequest(BaseModel):
+    auth_date: int
+    first_name: str
+    hash: str
+    id: int
+    photo_url: str

schemas/auth/responses.py

@@ -0,0 +1,6 @@
+from pydantic import BaseModel
+
+
+class AuthLoginResponse(BaseModel):
+    jwt_token: str
+    ok: bool

settings.py

@@ -10,3 +10,6 @@ PG_PASSWORD = os.environ.get('PG_PASSWORD')
 PG_PORT = os.environ.get('PG_PORT')
 PG_DATABASE = os.environ.get('PG_DATABASE')
 PG_HOST = os.environ.get('PG_HOST')
+
+# Telegram
+TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN')