blob: 7ae30471612e79dbfbb9660bf8836b4bbf035488 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
import os.path
import sqlite3
from pathlib import Path
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import FastAPI, Depends, Request
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from .config import Config
sq_con = None
@asynccontextmanager
async def lifespan(app: FastAPI):
db_dir = os.path.dirname(Config.DATABASE_PATH)
Path(db_dir).mkdir(parents=True, exist_ok=True)
sq_con = sqlite3.connect(Config.DATABASE_PATH)
yield
app = FastAPI(lifespan=lifespan)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
user = form_data.username
return {"access_token": user, "token_type": "bearer"}
@app.get("/{callsign}/tasks")
async def get_tasks(callsign: str, token: Annotated[str, Depends(oauth2_scheme)]):
return f'{{"callsign": "{callsign}", "token": "{token}"}}'
# if __name__ == "__main__":
# with open(os.path.join(os.path.dirname(__file__), Config.TOKEN_FILE_NAME)) as f:
# token = f.read().strip()
#
# d = daemon.Daemon(Config.AGENT_SYMBOL, token)
# d.run()
|
