Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@appier.route("/async/sender", "GET")
async def sender(self):
import asyncio
sleep = self.field("sleep", 1.0, cast = float)
self.request.set_content_type("text/plain")
await asyncio.sleep(sleep)
await self.request.send(b"Sender (1)\n")
await asyncio.sleep(sleep)
await self.request.send(b"Sender (2)\n")
await asyncio.sleep(sleep)
await self.request.send(b"Sender (3)\n")
@appier.route("/hello/", "GET")
def hello_count(self, count):
return dict(
message = "hello world %d" % count
)
@appier.route("/async/file", "GET")
def file(self):
file_path = self.field("path", None, mandatory = True)
delay = self.field("delay", 0.0, cast = float)
thread = self.field("thread", False, cast = bool)
type, _encoding = mimetypes.guess_type(file_path, strict = True)
type = type or "application/octet-stream"
self.request.content_type = type
for value in appier.header_a(): yield value
for value in appier.ensure_a(
self.read_file,
args = [file_path],
kwargs = dict(delay = delay),
thread = thread
): yield value
@appier.route("/hello", "GET")
def hello(self):
return dict(
message = "hello world"
)
@appier.route("/async/http", "GET")
async def http(self):
url = self.field("url", "https://www.flickr.com/")
delay = self.field("delay", 0.0, cast = float)
self.request.content_type = "text/html"
await appier.sleep(delay)
yield await appier.get_w(url)
@appier.route("/async/hello", "GET")
def hello(self):
partial = self.field("partial", True, cast = bool)
handler = self.handler_partial if partial else self.handler
yield from appier.header_a()
yield "before\n"
yield from handler()
yield "after\n"
@appier.route("/async/create", ("GET", "POST"))
async def create_(self):
name = self.field("name", "John Doe")
person = Person(name = name)
await person.save_a()
person = await person.reload_a(map = True)
return json.dumps(person)
@appier.route("/hello.tpl", "GET")
def hello_template(self):
return self.template(
"hello.txt",
message = "hello world"
)
@appier.route("/async/file", "GET")
async def file(self):
file_path = self.field("path", None, mandatory = True)
delay = self.field("delay", 0.0, cast = float)
thread = self.field("thread", False, cast = bool)
type, _encoding = mimetypes.guess_type(file_path, strict = True)
type = type or "application/octet-stream"
self.request.content_type = type
await appier.ensure_a(
self.read_file,
args = [file_path],
kwargs = dict(delay = delay),
thread = thread
)