@app.post("/")
async def test(request):
async def sample_streaming_fn(response):
await response.write("\nfoo 1")
await asyncio.sleep(1)
await response.write("\nbar 22")
await asyncio.sleep(1)
await response.write("\nfoo 333")
await asyncio.sleep(1)
await response.write("\nbar 4444")
response.protocol.close()
return stream(sample_streaming_fn, chunked=False)
response.protocol.close()
is what you are looking for.
KEEP_ALIVE
import uvloop
import asyncio
import logging
from functools import partial
from threading import Thread
from conf import settings
from .app import app
class ApiService:
def __init__(self, host, port):
self.host = host
self.port = port
self.__server = None
self.__loop = None
def run(self):
self.__loop = uvloop.new_event_loop()
thread = Thread(target=partial(self._thread_main, self.__loop), name="api-service")
thread.setDaemon(True)
thread.start()
def _thread_main(self, loop):
logging.info("ApiService Listening on http://{}:{}".format(self.host, self.port))
asyncio.set_event_loop(loop)
serv_coro = app.create_server(
host=self.host, port=self.port, return_asyncio_server=True, debug=settings.DEBUG
)
serv_task = asyncio.ensure_future(serv_coro, loop=self.__loop)
self.__server = self.__loop.run_until_complete(serv_task)
self.__server.after_start()
self.__loop.run_forever()
def stop(self):
if self.__loop is None or self.__server is None:
return
self.__loop.run_coroutine_threadsafe(self.__loop.stop)
self.__server.before_stop()
# Wait for server to close
close_task = self.__server.close()
self.__loop.run_until_complete(close_task)
# Complete all tasks on the loop
for connection in self.__server.connections:
connection.close_if_idle()
self.__server.after_stop()
Here is my code.
Hey Guys!
I have a question!
I`ve used sanic_session, but I have some routes that do not need to check session, like send_code or register or login
how can I prevent running this middleware code?
async def save_session(request, response):
"""After each request save the session, pass
the response to set client cookies.
"""
await self.interface.save(request, response)
and please help me in this issue: I want to use two database in my application, is it ok to have session_ids in both postgres (in sessions table) and in redis?
and when user registeres, the session id that I generated save in redis and postgres and redis is responsible to manage them?
so that in each request if there is the same session in redis, the user can do anything, but if not user can't access and direct to login page
request
object
@app.middleware("response")
async def save_session(request, response):
print(f"{request.path=}")
print(f"{request.endpoint=}")
app = Sanic("__BASE__")
bp = Blueprint("session stuff")
async def fake_save(*args):
print(f"Saving... {args=}")
@bp.middleware("response")
async def save_session(request, response):
await fake_save(request, response)
@app.get("/nosession")
async def nosession(request):
return text("No session.")
@bp.get("/session")
async def session(request):
return text("Yes session.")
app.blueprint(bp)
cool 👍app = Sanic("__BASE__") bp = Blueprint("session stuff") async def fake_save(*args): print(f"Saving... {args=}") @bp.middleware("response") async def save_session(request, response): await fake_save(request, response) @app.get("/nosession") async def nosession(request): return text("No session.") @bp.get("/session") async def session(request): return text("Yes session.") app.blueprint(bp)
app.websocket("/ws")
async def mws(request, ws):
while True:
...
r = app.add_task( heavyTask(x) )
if r:
...