add test file

app/auth.py

@@ -0,0 +1,80 @@
+from datetime import datetime, timedelta
+from typing import Optional
+
+from fastapi import Depends, HTTPException, status
+from jose import jwt, JWTError
+from fastapi.security import OAuth2PasswordBearer
+import bcrypt
+
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
+
+from app.models import User
+from app.schemas import Token
+from app.db import get_db
+import os
+
+# Загружаем переменные окружения
+SECRET_KEY = os.getenv("SECRET_KEY")
+ALGORITHM = os.getenv("ALGORITHM", "HS256")
+ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))
+
+# ---------------------------------------------------------
+# Хэширование пароля
+# ---------------------------------------------------------
+def hash_password(password: str) -> str:
+    pwd_bytes = password.encode('utf-8')
+    salt = bcrypt.gensalt()
+    return bcrypt.hashpw(pwd_bytes, salt).decode('utf-8')
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return bcrypt.checkpw(
+        plain_password.encode('utf-8'), 
+        hashed_password.encode('utf-8')
+    )
+
+# ---------------------------------------------------------
+# Создание JWT-токена
+# ---------------------------------------------------------
+def create_access_token(data: dict, expires_delta: Optional[int] = None):
+    to_encode = data.copy()
+
+    expire = datetime.utcnow() + timedelta(
+        minutes=expires_delta or ACCESS_TOKEN_EXPIRE_MINUTES
+    )
+    to_encode.update({"exp": expire})
+
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+
+# ---------------------------------------------------------
+# Получение текущего пользователя по токену
+# ---------------------------------------------------------
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+async def get_current_user(
+    token: str = Depends(oauth2_scheme),
+    db: AsyncSession = Depends(get_db)
+):
+    """ Достаём токен из Authorization Bearer """
+    if not token or not isinstance(token, str):
+        raise HTTPException(status_code=401, detail="Missing token")
+
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        email: str = payload.get("sub")
+
+        if email is None:
+            raise HTTPException(status_code=401, detail="Invalid token payload")
+
+    except JWTError:
+        raise HTTPException(status_code=401, detail="Invalid or expired token")
+
+    query = select(User).where(User.email == email)
+    result = await db.execute(query)
+    user = result.scalar_one_or_none()
+
+    if user is None:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    return user

app/core/config.py

@@ -0,0 +1,12 @@
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    # Секретный ключ — в реальном проекте брать из .env!
+    SECRET_KEY: str = "your-super-secret-key-change-this-in-production"
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
+
+    class Config:
+        env_file = ".env"
+
+settings = Settings()

app/db.py

@@ -0,0 +1,24 @@
+from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+from sqlalchemy.orm import sessionmaker, declarative_base
+import os
+from dotenv import load_dotenv
+
+
+load_dotenv()  # загрузить .env при старте
+
+
+DATABASE_URL = os.getenv("DATABASE_URL", "sqlite+aiosqlite:///./test.db")
+
+
+engine = create_async_engine(DATABASE_URL, echo=False, future=True)
+AsyncSessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
+
+
+Base = declarative_base()
+
+
+# Для удобства: зависимость (async) для получения сессии
+async def get_db():
+   async with AsyncSessionLocal() as session:
+       yield session
+

app/deps.py

@@ -0,0 +1,9 @@
+from fastapi import Depends, HTTPException
+from app.auth import get_current_user
+from app.models import User
+
+
+async def require_admin(user: User = Depends(get_current_user)):
+    if user.role != "admin":
+        raise HTTPException(status_code=403, detail="Admin access required")
+    return user

app/main.py

@@ -1,9 +1,73 @@
-from fastapi import FastAPI
+from fastapi import FastAPI, Depends, HTTPException, status
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.future import select
 
-# Создаем само приложение
+from app.db import engine, Base, get_db
-app = FastAPI()
+from app.models import User
+from app.schemas import UserCreate, UserOut, Token
+from app.auth import hash_password, verify_password, create_access_token
 
-# Вешаем обработчик на главную страницу
+app = FastAPI(title="FastAPI JWT Auth Example")
-@app.get("/")
+
-def read_root():
+
-    return {"message": "Hello, world!"}
+# ---------------------------------------------------------
+# Создать таблицы при запуске (для SQLite)
+# ---------------------------------------------------------
+@app.on_event("startup")
+async def startup():
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+
+
+# ---------------------------------------------------------
+# Регистрация
+# ---------------------------------------------------------
+@app.post("/register", response_model=UserOut)
+async def register(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
+    query = select(User).where(User.email == user_in.email)
+    result = await db.execute(query)
+    existing_user = result.scalar_one_or_none()
+
+    if existing_user:
+        raise HTTPException(status_code=400, detail="Email already exists")
+
+    user = User(
+        email=user_in.email,
+        hashed_password=hash_password(user_in.password),
+        role=user_in.role
+    )
+    db.add(user)
+    await db.commit()
+    await db.refresh(user)
+
+    return user
+
+
+# ---------------------------------------------------------
+# Логин
+# ---------------------------------------------------------
+@app.post("/login", response_model=Token)
+async def login(form: UserCreate, db: AsyncSession = Depends(get_db)):
+    query = select(User).where(User.email == form.email)
+    result = await db.execute(query)
+    user = result.scalar_one_or_none()
+
+    if not user:
+        raise HTTPException(status_code=401, detail="Invalid credentials")
+
+    if not verify_password(form.password, user.hashed_password):
+        raise HTTPException(status_code=401, detail="Invalid credentials")
+
+    token = create_access_token(data={"sub": user.email})
+
+    return {"access_token": token, "token_type": "bearer"}
+
+
+# ---------------------------------------------------------
+# Защищённый маршрут
+# ---------------------------------------------------------
+from app.auth import get_current_user
+
+@app.get("/me", response_model=UserOut)
+async def read_me(user=Depends(get_current_user)):
+    return user

app/models.py

@@ -0,0 +1,13 @@
+from sqlalchemy import Column, Integer, String, Boolean
+from app.db import Base
+
+
+class User(Base):
+   __tablename__ = "users"
+
+   id = Column(Integer, primary_key=True, index=True)
+   email = Column(String, unique=True, index=True, nullable=False)
+   hashed_password = Column(String, nullable=False)
+   is_active = Column(Boolean(), default=True)
+   role = Column(String, default="user")  # roles: "user", "admin"
+

app/schemas.py

@@ -0,0 +1,23 @@
+from pydantic import BaseModel, EmailStr
+
+
+class UserCreate(BaseModel):
+   email: EmailStr
+   password: str
+   role: str = "user"
+
+
+class UserOut(BaseModel):
+   id: int
+   email: EmailStr
+   is_active: bool
+   role: str
+
+
+   class Config:
+       orm_mode = True
+
+
+class Token(BaseModel):
+   access_token: str
+   token_type: str

app/schemas/user.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel, EmailStr
+
+class UserBase(BaseModel):
+    username: str
+    email: EmailStr
+
+class UserCreate(UserBase):
+    password: str
+
+class User(UserBase):
+    id: int
+    is_active: bool = True
+
+    class Config:
+        from_attributes = True
+
+class UserInDB(User):
+    hashed_password: str

requirements.txt

@@ -1 +1,6 @@
-fastapi[standard]
+fastapi[standard]
+sqlalchemy
+aiosqlite
+python-jose[cryptography]
+passlib[bcrypt]
+python-dotenv

testJWT.py

@@ -0,0 +1,30 @@
+import requests
+
+def register():
+  payload = {
+      "email": "tavres@example.com",
+      "password": "12345678",
+      "role": "superuser"
+  }
+  responce = requests.post('http://127.0.0.1:8000/register',json=payload)
+  return responce.text
+
+def login():
+  payload = {
+      "email": "user1@example.com",
+      "password": "12345678"
+  }
+  responce = requests.post('http://127.0.0.1:8000/login',json=payload)
+  return responce.json()
+
+def get_data(token):
+  headers = {
+    'Authorization':f'Bearer {token}',
+    'Content-Type': 'application/json'
+  }
+  print(headers)
+  responce = requests.get('http://127.0.0.1:8000/me',headers=headers)
+  return responce.text
+#print(register())
+data = login()
+print(get_data(data['access_token']))