Везде пишут, что aiohttp позволяет запускать асинхронные воркеры и при этом писать код так, будто бы он синхронный.
поэтому event loop перейдёт сразу к присваиванию.
import asyncio
import aioredis
from tornado import web, websocket
from tornado.ioloop import IOLoop
connections = []
class WSHandler(websocket.WebSocketHandler):
def open(self):
connections.append(self)
def on_message(self, message):
...
def on_close(self):
connections.remove(self)
class GetHandler(web.RequestHandler):
def get(self):
self.render("chat.html")
async def consumer(channel):
while await channel.wait_message():
msg = await channel.get(encoding='utf-8')
for connection in connections:
await connection.write_message(msg)
async def setup():
connection = await aioredis.create_redis('redis://localhost')
channel = await connection.subscribe('notifications')
asyncio.ensure_future(consumer(channel))
application = web.Application([
(r'/', GetHandler),
(r'/chat/', WSHandler),
])
if __name__ == '__main__':
application.listen(8000)
loop = IOLoop.current()
loop.add_callback(setup)
loop.start()
import ast
import sys
class DumbLisp(ast.NodeTransformer):
def fix(self, source, result):
ast.copy_location(result, source)
ast.fix_missing_locations(result)
return result
def visit_Expr(self, node):
if isinstance(node.value, ast.Tuple):
t = node.value
if isinstance(t.elts[0], ast.Name):
if t.elts[0].id == 'newif':
_else = [ast.Expr(value=t.elts[3])] if len(t.elts) == 4 else []
_if = ast.If(test=t.elts[1],
body=[ast.Expr(value=t.elts[2])],
orelse=_else)
self.generic_visit(_if)
return self.fix(node, _if)
elif t.elts[0].id == 'define':
assign = ast.Assign(targets=[ast.Name(id=t.elts[1].id, ctx=ast.Store())],
value=t.elts[2])
return self.fix(node, assign)
else:
call = ast.Expr(value=ast.Call(func=t.elts[0], args=t.elts[1:], keywords=[]))
return self.fix(node, call)
return node
with open(sys.argv[1]) as fh:
tree = ast.parse(fh.read())
DumbLisp().visit(tree)
code = compile(tree, filename=sys.argv[1], mode="exec")
exec(code)
(define, a, 1)
(newif, a == 1, (print, 'Yes'), (print, 'No'))
python dumb_translator.py test.dl
согласованное состояние восстановления достигнуто
начало передачи журнала с главного сервера, с позиции 4A4/E000000 на линии времени 2
@admin.register(CompanyNews)
class CompanyNewsAdmin(admin.ModelAdmin):
...
def get_fields(self, request, obj=None):
fields = super(StudentAdmin, self).get_fields(request, obj)
if not request.user.is_superuser:
fields.remove('IsVisible')
return fields
и недавно узнал, что html шаблоны можно кодировать в base64 и хранить в бд MySQLа теперь забудьте.
class MyConsumer(AsyncWebsocketConsumer):
async def disconnect(self, close_code):
# соединение закрылось
from threading import Thread
from django.core.mail import send_mail
Thread(target=send_mail, args=(subject, txt_msg_body, settings.DEFAULT_FROM_EMAIL, smtp_to),
kwargs={'fail_silently': (not settings.DEBUG), 'html_message': html_msg_body}).start()
from datetime import timedelta
from django.db import models
from django.core.exceptions import ValidationError
from django.utils import timezone
class Person(models.Model):
dob = models.DateField('Дата рождения')
def clean_fields(self, exclude=None):
super().clean_fields(exclude=exclude)
now = timezone.now()
if self.dob > (now - timedelta(days=30)):
raise ValidationError('Рано ещё заносить в базу')
if (now.year - self.dob.year) > 120:
raise ValidationError('Поздно уже заносить в базу')