-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
97 lines (72 loc) · 2.98 KB
/
main.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
import warnings
from typing import List
from py_api.plugins.plugin_helper import PluginLoader, PluginSchema
from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles
from py_api.routers import routers_user, routers_auth, routers_client, routers_event, routers_contact, routers_todo, \
routers_invitation
from tortoise.contrib.fastapi import register_tortoise
from py_api.config import Config
Config.get_instance().register_needed_keys(keys=['JWT_REFRESH_TOKEN_SECRET', 'JWT_ACCESS_TOKEN_SECRET', 'SYSTEM_ID'])
models = ['py_api.models.models_user', 'py_api.models.models_todo', 'py_api.models.models_event',
'py_api.models.models_contact']
plugin_loader = PluginLoader()
plugin_loader.load_plugins()
print_text = 'Plugin {} could not be loaded because {}'
routers_to_add = {}
for plugin in plugin_loader.plugins:
routers = plugin.serve_routers()
correct_routers = True
for router in routers:
if not router.prefix.startswith(f'/plugin/{plugin.name}'):
correct_routers = False
break
if not correct_routers:
warnings.warn(print_text.format(plugin.name, 'the router is not correctly defined'))
plugin_loader.plugins.remove(plugin)
continue
routers_to_add[plugin.name] = routers
models.append(*plugin.models)
if plugin.needed_env_keys is not None and plugin.needed_env_keys is not []:
Config.get_instance().register_needed_keys(keys=plugin.needed_env_keys)
Config.get_instance().validate_needed_keys()
app = FastAPI()
app.include_router(routers_auth.router)
app.include_router(routers_user.router)
app.include_router(routers_client.router)
app.include_router(routers_event.router)
app.include_router(routers_contact.router)
app.include_router(routers_todo.router)
app.include_router(routers_invitation.router)
app.mount("/static", StaticFiles(directory="static"), name="static")
for plugin in plugin_loader.plugins:
for router in routers_to_add[plugin.name]:
app.include_router(router)
if plugin.has_static_files:
app.mount(f'/{plugin.name}', StaticFiles(directory=f'py_api/plugins/{plugin.name}/static'), name=plugin.name)
@app.get("/plugins", response_model=List[PluginSchema])
def get_plugins() -> List[PluginSchema]:
plugin_list = []
for plugin in plugin_loader.plugins:
plugin_list.append(PluginSchema(
name=plugin.name,
routes=[router.prefix for router in routers_to_add[plugin.name]]
))
return plugin_list
@app.get("/hello-world")
def hello():
return {
'success': True,
'message': 'Hello World'
}
@app.get("/favicon.ico", response_class=FileResponse, include_in_schema=False)
async def favicon():
return FileResponse('static/img/favicon.ico')
register_tortoise(
app,
db_url=Config.get_instance().get_config_value('DATABASE_URL', 'sqlite://files/db.sqlite3'),
modules={'models': models},
generate_schemas=True,
add_exception_handlers=True,
)