First commit

Author: michqo

.editorconfig

@@ -0,0 +1,23 @@
+root = true
+
+# Unix-style newlines with a newline ending every file
+[*]
+insert_final_newline = true
+trim_trailing_whitespace = true
+end_of_line = lf
+charset = utf-8
+
+# https://www.python.org/dev/peps/pep-0008/
+[*.py]
+indent_style = space
+indent_size = 4
+max_line_length = 119
+
+[*.html]
+indent_style = tab
+indent_size = 2
+
+[*.md]
+max_line_length = 119
+indent_size = 4
+indent_style = space

.env

@@ -0,0 +1 @@
+SQLALCHEMY_DATABASE_URL = 'postgresql://postgres:root@localhost/database_backup'

.env.example

@@ -0,0 +1 @@
+DATABASE_URI = 'postgresql://postgres:<password>@localhost/<name_of_the_datbase>'

.gitignore

@@ -0,0 +1,25 @@
+.idea
+.ipynb_checkpoints
+.mypy_cache
+.vscode
+__pycache__
+.pytest_cache
+htmlcov
+dist
+site
+.coverage
+coverage.xml
+.netlify
+test.db
+log.txt
+Pipfile.lock
+env3.*
+env
+docs_build
+venv
+docs.zip
+archive.zip
+
+# vim temporary files
+*~
+.*.sw?

app/__init__.py

@@ -0,0 +1,13 @@
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+    SQLALCHEMY_DATABASE_URL: str
+    APP_NAME: str = "DB Backup API"
+
+    class Config:
+        case_sensitive = True
+        env_file = ".env"
+
+
+settings = Settings()  # type: ignore

app/config.py

@@ -0,0 +1,15 @@
+from sqlalchemy.orm import Session
+
+from . import models, schemas
+
+
+def create_user(db: Session, user: schemas.User):
+    db_user = models.User(username=user.username, password=user.password)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+
+def get_user(db: Session, username: str):
+    return db.query(models.User).filter(models.User.username == username).first()

app/crud.py

@@ -0,0 +1,10 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+from .config import settings
+
+engine = create_engine(settings.SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

app/database.py

@@ -0,0 +1,21 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from .router import router
+
+app = FastAPI()
+
+origins = [
+    "http://localhost:8080",
+    "http://localhost:5173",
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(router)

app/main.py

@@ -0,0 +1,10 @@
+from sqlalchemy import Column, String, Integer
+
+from .database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+    id = Column(Integer, primary_key=True)
+    username = Column(String)
+    password = Column(String)

app/models.py

@@ -0,0 +1,113 @@
+from datetime import datetime, timedelta
+from typing import Annotated
+from jose import JWTError, jwt
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+
+from . import crud, schemas
+from . import models
+from .database import SessionLocal, engine
+from passlib.context import CryptContext
+
+# to get a string like this run:
+# openssl rand -hex 32
+SECRET_KEY = "89afa0ea05e272f5a746f466be7c256d982d307903e8201ad8f6c11450e71d7f"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 24 * 60
+access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+
+
+models.Base.metadata.create_all(bind=engine)
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+
+router = APIRouter()
+
+
+# Dependency
+"""
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+"""
+
+db = SessionLocal()
+
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+
+def create_access_token(data: dict, expires_delta: timedelta | None = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def authenticate_user(username: str, password: str):
+    user = crud.get_user(db, username)
+    if not user:
+        return False
+    if not verify_password(password, user.password):
+        return False
+    return user
+
+
+async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username = payload.get("sub")
+        if username is None:
+            raise credentials_exception
+        token_data = schemas.TokenData(username=username)
+    except JWTError:
+        raise credentials_exception
+    user = crud.get_user(db, token_data.username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+
+@router.post("/token")
+async def login_for_access_token(
+    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
+):
+    user = authenticate_user(form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect username or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+    access_token = create_access_token(
+        data={"sub": user.username}, expires_delta=access_token_expires
+    )
+    return {"access_token": access_token, "token_type": "bearer"}
+
+
+@router.get("/users/me")
+async def read_users_me(
+    current_user: Annotated[schemas.User, Depends(get_current_user)]
+):
+    return current_user.username

app/router.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel
+
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+
+class TokenData(BaseModel):
+    username: str
+
+
+class User(BaseModel):
+    username: str
+    password: str
+
+    class Config:
+        orm_mode = True