Init

.gitignore

@@ -0,0 +1,8 @@
+venv/
+.idea/
+*.pyc
+__pycache__
+*.egg-info/
+build/
+dist/
+*.egg

main.py

@@ -0,0 +1,77 @@
+from datetime import datetime, timedelta
+from pydantic import BaseModel
+
+import asyncpg
+from fastapi import FastAPI, Depends, HTTPException
+import jwt
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+
+SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
+POSTGRES_URL = 'postgresql://amonic:amonic123@92.53.120.110:5432/amonic'
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+app = FastAPI(title="AMONIC API", version='0.0.1')
+db: asyncpg.connection.Connection | None = None
+
+
+class UserDB(BaseModel):
+    id: int | None
+    roleid: int | None
+    email: str | None
+    password: str | None
+    firstname: str | None
+    lastname: str | None
+    officeid: int | None
+    birthdate: datetime | None
+    active: bool | None
+
+
+@app.on_event('startup')
+async def startup():
+    global db
+    db = await asyncpg.connect(POSTGRES_URL)
+
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/login")
+
+
+async def get_current_user(token: str = Depends(oauth2_scheme)):
+    token = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+    sql = """
+    select * from users where email= $1
+    """
+    row = await db.fetchrow(sql, token['sub'])
+    if row is None:
+        raise HTTPException(
+            status_code=404,
+            detail="User not found"
+        )
+    return UserDB.model_validate(dict(row))
+
+
+@app.post('/login')
+async def login(user_data: OAuth2PasswordRequestForm = Depends()):
+    sql = """
+    select * from users where email= $1 and password = md5($2)
+    """
+    row = await db.fetchrow(sql, user_data.username, user_data.password)
+    if row is None:
+        raise HTTPException(
+            status_code=401,
+            detail="Incorrect username or password"
+        )
+    payload = {
+        "sub": user_data.username,
+        "exp": datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    }
+    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
+    return {"access_token": token, "token_type": "bearer"}
+
+
+@app.get('/current-user')
+async def profiles(user: UserDB = Depends(get_current_user)):
+    return user
+
+# @app.get('/profiles')
+# async def ():