|
import pathlib
|
|
|
|
from aiohttp import web
|
|
from aiohttp_session import (
|
|
get_session as session_get,
|
|
session_middleware,
|
|
setup as session_setup,
|
|
)
|
|
from aiohttp_session.cookie_storage import EncryptedCookieStorage
|
|
from app.app import Application as BaseApplication
|
|
|
|
from snek.docs.app import Application as DocsApplication
|
|
from snek.mapper import get_mappers
|
|
from snek.service import get_services
|
|
from snek.system import http
|
|
from snek.system.cache import Cache
|
|
from snek.system.markdown import MarkdownExtension
|
|
from snek.system.middleware import cors_middleware
|
|
from snek.view.about import AboutHTMLView, AboutMDView
|
|
from snek.view.docs import DocsHTMLView, DocsMDView
|
|
from snek.view.index import IndexView
|
|
from snek.view.login import LoginView
|
|
from snek.view.logout import LogoutView
|
|
from snek.view.register import RegisterView
|
|
from snek.view.rpc import RPCView
|
|
from snek.view.status import StatusView
|
|
from snek.view.web import WebView
|
|
|
|
# base64.urlsafe_b64encode(
|
|
SESSION_KEY = b"c79a0c5fda4b424189c427d28c9f7c34"
|
|
|
|
|
|
@web.middleware
|
|
async def session_middleware(request, handler):
|
|
setattr(request, "session", await session_get(request))
|
|
response = await handler(request)
|
|
return response
|
|
|
|
|
|
class Application(BaseApplication):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
middlewares = [
|
|
cors_middleware,
|
|
web.normalize_path_middleware(merge_slashes=True),
|
|
]
|
|
self.template_path = pathlib.Path(__file__).parent.joinpath("templates")
|
|
super().__init__(
|
|
middlewares=middlewares, template_path=self.template_path, *args, **kwargs
|
|
)
|
|
session_setup(self, EncryptedCookieStorage(SESSION_KEY))
|
|
self._middlewares.append(session_middleware)
|
|
self.jinja2_env.add_extension(MarkdownExtension)
|
|
self.setup_router()
|
|
self.cache = Cache(self)
|
|
self.services = get_services(app=self)
|
|
self.mappers = get_mappers(app=self)
|
|
|
|
def setup_router(self):
|
|
self.router.add_get("/", IndexView)
|
|
self.router.add_static(
|
|
"/",
|
|
pathlib.Path(__file__).parent.joinpath("static"),
|
|
name="static",
|
|
show_index=True,
|
|
)
|
|
self.router.add_view("/about.html", AboutHTMLView)
|
|
self.router.add_view("/about.md", AboutMDView)
|
|
self.router.add_view("/logout.json", LogoutView)
|
|
self.router.add_view("/logout.html", LogoutView)
|
|
self.router.add_view("/docs.html", DocsHTMLView)
|
|
self.router.add_view("/docs.md", DocsMDView)
|
|
self.router.add_view("/status.json", StatusView)
|
|
self.router.add_view("/web.html", WebView)
|
|
self.router.add_view("/login.html", LoginView)
|
|
self.router.add_view("/login.json", LoginView)
|
|
self.router.add_view("/register.html", RegisterView)
|
|
self.router.add_view("/register.json", RegisterView)
|
|
self.router.add_get("/http-get", self.handle_http_get)
|
|
self.router.add_get("/http-photo", self.handle_http_photo)
|
|
self.router.add_get("/rpc.ws", RPCView)
|
|
self.add_subapp(
|
|
"/docs",
|
|
DocsApplication(path=pathlib.Path(__file__).parent.joinpath("docs")),
|
|
)
|
|
|
|
async def handle_test(self, request):
|
|
|
|
return await self.render_template(
|
|
"test.html", request, context={"name": "retoor"}
|
|
)
|
|
|
|
async def handle_http_get(self, request: web.Request):
|
|
url = request.query.get("url")
|
|
content = await http.get(url)
|
|
return web.Response(body=content)
|
|
|
|
async def handle_http_photo(self, request):
|
|
url = request.query.get("url")
|
|
path = await http.create_site_photo(url)
|
|
return web.Response(
|
|
body=path.read_bytes(), headers={"Content-Type": "image/png"}
|
|
)
|
|
|
|
# @time_cache_async(60)
|
|
async def render_template(self, template, request, context=None):
|
|
return await super().render_template(template, request, context)
|
|
|
|
|
|
app = Application(db_path="sqlite:///snek.db")
|
|
|
|
if __name__ == "__main__":
|
|
|
|
web.run_app(app, port=8081, host="0.0.0.0")
|