lavkach3

1
import asyncio
2
import logging
3

4
from fastapi_restful.tasks import repeat_every
5
from pydantic import BaseModel
6
from pydantic.v1.schema import schema
7

8
from app.front.tkq import broker
9

10
logging.basicConfig(level=logging.INFO)
11
logger = logging.getLogger(__name__)
12

13
cursors = {}
14

15
@broker.task
16
async def import_prepare_data(model: str, data: list, ):
17
    env = broker.env
18
    model = env[model]
19
    import_schema: BaseModel
20
    header = list(data[0].keys())
21
    errors = []
22
    lines: list = []
23
    mode = 'create'
24
    if 'id' in header:
25
        import_schema = model.schemas.update
26
        mode = 'update'
27
    else:
28
        import_schema = model.schemas.create
29
    for i, line in enumerate(data):
30
        try:
31
            line = import_schema(**line)
32
        except Exception as ex:
33
            logger.error(f'Error in line {i}: {ex}')
34
            for er in ex.errors():
35
                errors.append({
36
                    'line': i+2,
37
                    'type': er.get('type'),
38
                    'msg': f"{er.get('msg')}: Input value: {er.get('input')}",
39
                    'field': str(er.get('loc')[0])
40
                })
41
            continue
42
        lines.append(line.model_dump(mode='json'))
43
    return errors, lines, mode
44

45
@broker.task
46
async def import_save(model: str, data: list):
47
    env = broker.env
48
    model = env[model]
49
    import_schema: BaseModel
50
    imported_lines: list = []
51
    async with model.adapter as a:
52
        for line in data:
53
            resp = await a.create(json=line, model=model.name)
54
            imported_lines.append(resp)
55
    return imported_lines