forked from osm-fr/osmose-frontend
-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
131 lines (104 loc) · 3.72 KB
/
app.py
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
from typing import Any, Dict, Optional
from uuid import UUID, uuid4
from fastapi import Depends, FastAPI, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.utils import get_openapi
from fastapi.responses import HTMLResponse, RedirectResponse
from modules import utils
from modules.dependencies import database
from . import byuser, editor, false_positive, issue, issues, map
from .tool import oauth, xmldict
from .tool.session import SessionData, backend, cookie, verifier
openapi_tags = [
{
"name": "Private API",
"description": "Private API. For external use refer to public API.",
},
]
app = FastAPI(openapi_tags=openapi_tags)
def custom_openapi() -> Dict[str, Any]:
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Osmose-QA API",
version="0.3",
routes=app.routes,
)
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi # type: ignore
@app.get("/login")
async def login(session_id: Optional[UUID] = Depends(cookie)) -> RedirectResponse:
if session_id:
await backend.delete(session_id)
(url, oauth_tokens) = oauth.fetch_request_token()
session = uuid4()
await backend.create(session, SessionData(oauth_tokens=oauth_tokens))
response = RedirectResponse(url)
cookie.attach_to_response(response, session)
return response
@app.get("/logout")
async def logout(
response: Response, session_id: Optional[UUID] = Depends(cookie)
) -> RedirectResponse:
if session_id:
await backend.delete(session_id)
cookie.delete_from_response(response)
return RedirectResponse("map/")
@app.get("/oauth")
async def oauth_(
session_id: UUID = Depends(cookie),
session_data: Optional[SessionData] = Depends(verifier),
) -> RedirectResponse:
if session_id and session_data:
try:
oauth_tokens = oauth.fetch_access_token(session_data.oauth_tokens)
session_data.oauth_tokens = oauth_tokens
user_request = oauth.get(
oauth_tokens, utils.remote_url + "api/0.6/user/details"
)
if user_request:
session_data.user = xmldict.xml_to_dict(user_request)
await backend.update(session_id, session_data)
except Exception:
pass
return RedirectResponse("map/")
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "PUT", "DELETE"],
)
@app.on_event("startup")
async def startup() -> None:
await database.startup()
# Add routes
app.include_router(byuser.router)
app.include_router(editor.router)
app.include_router(false_positive.router)
app.include_router(issue.router)
app.include_router(issues.router)
app.include_router(map.router)
@app.get("/", tags=["vue"])
@app.get("/contact", tags=["vue"])
@app.get("/copyright", tags=["vue"])
@app.get("/translation", tags=["vue"])
@app.get("/map/", tags=["vue"])
@app.get("/issue/{uuid}", tags=["vue"])
@app.get("/false-positive/{uuid}", tags=["vue"])
@app.get("/issues/open", tags=["vue"])
@app.get("/issues/done", tags=["vue"])
@app.get("/issues/false-positive", tags=["vue"])
@app.get("/issues/matrix", tags=["vue"])
@app.get("/byuser/", tags=["vue"])
@app.get("/byuser/{username}", tags=["vue"])
@app.get("/control/update/{source}", tags=["vue"])
@app.get("/control/update_matrix", tags=["vue"])
@app.get("/control/update_summary", tags=["vue"])
@app.get("/control/update_summary_by_analyser", tags=["vue"])
def vue(
uuid: Optional[UUID] = None,
username: Optional[str] = None,
source: Optional[str] = None,
) -> HTMLResponse:
return HTMLResponse(content=open("web/public/assets/index.html").read())