@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)
I think
response.protocol.close() is what you are looking for.
You can also just exit out of your loop. That would close it, eventually
It would stay open as long as your
KEEP_ALIVE
so, calling explicit closing may be what you are after.
1 reply
authentication?
Something that implements a blog posting api, handles image uploads and handles basic image manipulations. Something like the wagtail layer on top of django.
Headless CRMS seem to be the new rage at the moment now that I've been picking up some react (and indeed that react is pretty nice). I guess it's a well known space so was curious if there were any modules being developed for sanic already.
I might take a look at taking a well known architecture / model and just write a plugin.
Headless CRMS seem to be the new rage at the moment now that I've been picking up some react (and indeed that react is pretty nice). I guess it's a well known space so was curious if there were any modules being developed for sanic already.
I might take a look at taking a well known architecture / model and just write a plugin.
but close the server have a question
Traceback (most recent call last):
File "/Users/liuzhichao/liuzhichao/trunk/code/TrunkPort/TrunkPortGateway/main.py", line 65, in <module>
app.stop()
File "/Users/liuzhichao/liuzhichao/trunk/code/TrunkPort/TrunkPortGateway/app.py", line 40, in stop
self.api_service.stop()
File "/Users/liuzhichao/liuzhichao/trunk/code/TrunkPort/TrunkPortGateway/web/init.py", line 53, in stop
self.loop.run_until_complete(close_task)
File "uvloop/loop.pyx", line 1450, in uvloop.loop.Loop.run_until_complete
File "uvloop/loop.pyx", line 1443, in uvloop.loop.Loop.run_until_complete
File "uvloop/loop.pyx", line 1351, in uvloop.loop.Loop.run_forever
File "uvloop/loop.pyx", line 480, in uvloop.loop.Loop._run
RuntimeError: this event loop is already running.
File "/Users/liuzhichao/liuzhichao/trunk/code/TrunkPort/TrunkPortGateway/main.py", line 65, in <module>
app.stop()
File "/Users/liuzhichao/liuzhichao/trunk/code/TrunkPort/TrunkPortGateway/app.py", line 40, in stop
self.api_service.stop()
File "/Users/liuzhichao/liuzhichao/trunk/code/TrunkPort/TrunkPortGateway/web/init.py", line 53, in stop
self.loop.run_until_complete(close_task)
File "uvloop/loop.pyx", line 1450, in uvloop.loop.Loop.run_until_complete
File "uvloop/loop.pyx", line 1443, in uvloop.loop.Loop.run_until_complete
File "uvloop/loop.pyx", line 1351, in uvloop.loop.Loop.run_forever
File "uvloop/loop.pyx", line 480, in uvloop.loop.Loop._run
RuntimeError: this event loop is already running.
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()
let's continue in the thread in the forum
I don't know much about this project, other than it does exist
But, if it is not something that would interest you, it is very simple to build your own with the usage of middleware (see Sanic docs)
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
The easiest would be to simply check the
request object
@app.middleware("response")
async def save_session(request, response):
print(f"{request.path=}")
print(f"{request.endpoint=}")
Another would be to use blueprints
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)
both solutions are great, thank you so much 🙏
I am new to async and sanic, I have been googling but can't find a solution to this:
I am trying to put a heavy task under add_task, and get back the return tho can't find how to do it (or if its the way to do it, I may need to change the logic there):
I am trying to put a heavy task under add_task, and get back the return tho can't find how to do it (or if its the way to do it, I may need to change the logic there):
app.websocket("/ws")
async def mws(request, ws):
while True:
...
r = app.add_task( heavyTask(x) )
if r:
...