entity.py
2.69 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import time
from typing import List, Any, Optional, Union
from fastapi import APIRouter, Depends, Request, HTTPException, Body
from sqlalchemy.orm import Session
from sqlalchemy import and_, or_
from presupuestov1 import model, schema, constant, schemas, models, search
from aiocache import cached
router = APIRouter(
prefix='/entidad',
tags=['entidad'],
)
###############################################################################
# GET
###############################################################################
@router.get('/{entity_id:int}', response_model=schemas.classifier.ClasInstitucional)
def fetch_entity(
entity_id: int,
db: Session = Depends(model.get_db),
):
ClasInstitucional = models.classifier.ClasInstitucional
return db.query(ClasInstitucional).filter(
ClasInstitucional.entidad == entity_id
).one()
@router.get('/clasificador', response_model=List[schemas.classifier.ClasInstitucional])
def fetch_classifier(
db: Session = Depends(model.get_db),
):
return db.query(models.classifier.ClasInstitucional).all()
@router.get('/{entity_id:int}/objetos', response_model=List[schemas.treemap.TreemapObjeto])
def fetch_entity_objetos(
entity_id: int,
db: Session = Depends(model.get_db),
query: schema.QueryParams = Depends(),
):
TreemapObjeto = models.treemap.TreemapObjeto
return get_entity_connections(entity_id, db, query, TreemapObjeto, 'objeto')
router.get('/{entity_id:int}/finfuns', response_model=List[schemas.treemap.TreemapFinFun])
def fetch_entity_finfuns(
entity_id: int,
db: Session = Depends(model.get_db),
query: schema.QueryParams = Depends(),
):
TreemapFinFun = models.treemap.TreemapFinFun
return get_entity_connections(entity_id, db, query, TreemapFinFun, 'finfun')
@router.get('/{entity_id:int}/actecos', response_model=List[schemas.treemap.TreemapActeco])
def fetch_entity_actecos(
entity_id: int,
db: Session = Depends(model.get_db),
query: schema.QueryParams = Depends(),
):
TreemapActeco = models.treemap.TreemapActeco
return get_entity_connections(entity_id, db, query, TreemapActeco, 'acteco')
###############################################################################
# Private
###############################################################################
def get_entity_connections(entity_id, db, query, ClassObject, object_id):
if query.gestion is None:
query.gestion = 0
query_filters = [
ClassObject.entidad == entity_id,
ClassObject.gestion == query.gestion,
]
print(query.id is not None)
if query.id is not None:
ids = query.id.split(',')
ids = [int(_.strip()) for _ in ids]
ids = or_([getattr(ClassObject, object_id) == _ for _ in ids])
query_filters.append(ids)
return db.query(ClassObject).filter(and_(*query_filters)).all()