Infra
/
risotto
9
0
Fork 1
Code Issues 1 Releases Wiki Activity

Merge branch 'develop' into dist/risotto/risotto-2.8.0/develop

This commit is contained in:
Emmanuel Garette 2021-04-25 20:32:16 +02:00
parent 2a98575790 1063d2e735
commit c63170be1d
4 changed files with 191 additions and 176 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/risotto/context.py
View File

@ -3,3 +3,11 @@ class Context:
self.paths = []
self.context_id = None
self.start_id = None
def copy(self):
context = Context()
for key, value in self.__dict__.items():
if key.startswith('__'):
continue
setattr(context, key, value)
return context

53
src/risotto/dispatcher.py
View File

@ -89,7 +89,6 @@ class CallDispatcher:
if hasattr(old_risotto_context, 'connection'):
# do not start a new database connection
risotto_context.connection = old_risotto_context.connection
risotto_context.log_connection = old_risotto_context.log_connection
await log.start(risotto_context,
kwargs,
info_msg,
@ -119,14 +118,6 @@ class CallDispatcher:
raise CallError(err) from err
else:
error = None
async with self.pool.acquire() as log_connection:
await log_connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.log_connection = log_connection
try:
async with self.pool.acquire() as connection:
await connection.set_type_codec(
@ -198,8 +189,6 @@ class PublishDispatcher:
for message, message_infos in messages.items():
# event not emit locally
if message_infos['pattern'] == 'event' and 'functions' in message_infos and message_infos['functions']:
# module, submodule, submessage = message.split('.', 2)
# if f'{module}.{submodule}' not in self.injected_self:
uri = f'{version}.{message}'
print(f' - {uri}')
await self.listened_connection.add_listener(uri,
@ -235,21 +224,34 @@ class PublishDispatcher:
version, message = uri.split('.', 1)
loop = get_event_loop()
remote_kw = loads(payload)
for function_obj in self.messages[version][message]['functions']:
risotto_context = self.build_new_context(remote_kw['context'],
version,
message,
'event',
)
callback = lambda: ensure_future(self._publish(version,
message,
risotto_context,
**remote_kw['kwargs'],
))
callback = self.get_callback(version, message, function_obj, risotto_context, remote_kw['kwargs'],)
loop.call_soon(callback)
def get_callback(self,
version,
message,
function_obj,
risotto_context,
kwargs,
):
return lambda: ensure_future(self._publish(version,
message,
function_obj,
risotto_context,
**kwargs,
))
async def _publish(self,
version: str,
message: str,
function_obj,
risotto_context: Context,
**kwargs,
) -> None:
@ -259,15 +261,6 @@ class PublishDispatcher:
False,
False,
)
for function_obj in self.messages[version][message]['functions']:
async with self.pool.acquire() as log_connection:
await log_connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.log_connection = log_connection
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
@ -307,15 +300,6 @@ class PublishDispatcher:
# if there is a problem with arguments, log and do nothing
if get_config()['global']['debug']:
print_exc()
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
risotto_context.connection = connection
async with connection.transaction():
await log.failed(risotto_context,
str(err),
)
@ -346,6 +330,7 @@ class Dispatcher(register.RegisterDispatcher,
risotto_context.type = type
risotto_context.message = message
risotto_context.version = version
risotto_context.pool = self.pool
return risotto_context
async def check_message_type(self,

39
src/risotto/logger.py
View File

@ -2,15 +2,34 @@ from typing import Dict, Any, Optional
from json import dumps, loads
from asyncpg.exceptions import UndefinedTableError
from datetime import datetime
from asyncio import Lock
from .context import Context
from .utils import _
from .config import get_config
database_lock = Lock()
class Logger:
""" An object to manager log
"""
def __init__(self) -> None:
self.log_connection = None
async def get_connection(self,
risotto_context: Context,
):
if not self.log_connection:
self.log_connection = await risotto_context.pool.acquire()
await self.log_connection.set_type_codec(
'json',
encoder=dumps,
decoder=loads,
schema='pg_catalog'
)
return self.log_connection
async def insert(self,
msg: str,
risotto_context: Context,
@ -38,8 +57,9 @@ class Logger:
sql = insert + ') ' + values + ') RETURNING LogId'
try:
async with risotto_context.log_connection.transaction():
log_id = await risotto_context.log_connection.fetchval(sql, *args)
async with database_lock:
connection = await self.get_connection(risotto_context)
log_id = await connection.fetchval(sql, *args)
if context_id is None and start:
risotto_context.context_id = log_id
if start:
@ -61,8 +81,9 @@ class Logger:
sql += ' AND URI = $3'
args.append(uri)
ret = []
async with risotto_context.log_connection.transaction():
for row in await risotto_context.log_connection.fetch(*args):
async with database_lock:
connection = await self.get_connection(risotto_context)
for row in await connection.fetch(*args):
d = {}
for key, value in row.items():
if key in ['kwargs', 'returns']:
@ -173,8 +194,9 @@ class Logger:
args.append(dumps(returns))
sql += """WHERE LogId = $1
"""
async with risotto_context.log_connection.transaction():
await risotto_context.log_connection.execute(sql,
async with database_lock:
connection = await self.get_connection(risotto_context)
await connection.execute(sql,
risotto_context.start_id,
*args,
)
@ -196,8 +218,9 @@ class Logger:
Msg = $3
WHERE LogId = $1
"""
async with risotto_context.log_connection.transaction():
await risotto_context.log_connection.execute(sql,
async with database_lock:
connection = await self.get_connection(risotto_context)
await connection.execute(sql,
risotto_context.start_id,
datetime.now(),
err,

3
src/risotto/register.py
View File

@ -297,7 +297,6 @@ class RegisterDispatcher:
truncate: bool=False,
) -> None:
internal_user = get_config()['global']['internal_user']
async with self.pool.acquire() as log_connection:
async with self.pool.acquire() as connection:
await connection.set_type_codec(
'json',
@ -314,7 +313,7 @@ class RegisterDispatcher:
risotto_context.username = internal_user
risotto_context.paths.append(f'internal.{submodule_name}.on_join')
risotto_context.type = None
risotto_context.log_connection = log_connection
risotto_context.pool = self.pool
risotto_context.connection = connection
risotto_context.module = submodule_name.split('.', 1)[0]
info_msg = _(f'in function risotto_{submodule_name}.on_join')