daily.py
4.26 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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
from typing import List
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session, aliased
from sqlalchemy import func
from presupuestov1 import model, schema, schemas, models
from presupuestov1.routers._decorators import not_found
router = APIRouter(
prefix='/diario',
tags=['diario'],
)
###############################################################################
# GET
###############################################################################
@router.get('/status', response_model=schemas.daily.UpdateStatus)
@not_found
def fetch_status(
db: Session = Depends(model.get_db),
):
Daily = models.daily.UpdateStatus
return db.query(Daily).order_by(
Daily.fecha_actualizacion.desc()
).first()
@router.get('/clasificadores', response_model=List[schemas.daily.EjecucionClasificadores])
@not_found
def fetch_clasificadores(
db: Session = Depends(model.get_db),
query: schema.DailyParams = Depends(),
):
Daily = models.daily.EjecucionClasificadores
filters = []
if query.transferencias is not None:
filters.append(Daily.transferencias == query.transferencias)
if query.dias is not None:
filters.append(Daily.dias == query.dias)
return db.query(Daily).filter(*filters).order_by(Daily.devengado.desc()).all()
@router.get('/entidades', response_model=List[schemas.daily.EjecucionEntidades])
@not_found
def fetch_entidades(
db: Session = Depends(model.get_db),
query: schema.DailyParams = Depends(),
):
Daily = models.daily.EjecucionEntidades
transferencias = query.transferencias or False # :O
page = query.page or 1
page_size = 10
filters = []
if not transferencias:
filters.append(Daily.transferencias == query.transferencias)
if query.dias is not None:
filters.append(Daily.dias == query.dias)
if query.entidad is not None:
filters.append(Daily.entidad == query.entidad)
return db.query(Daily).filter(*filters).order_by(Daily.devengado.desc()).offset(
(page - 1) * page_size
).limit(page_size).all()
@router.get('/timeline', response_model=List[schemas.daily.Timeline])
@not_found
def fetch_timeline(
db: Session = Depends(model.get_db),
query: schema.DailyParams = Depends(),
):
Daily = models.daily.Timeline
page = query.page or 1
page_size = 10
if query.entidad is not None:
stmt = (
select(Daily)
.where(Daily.entidad == query.entidad)
.order_by(Daily.devengado.desc(), Daily.id)
.offset((page - 1) * page_size)
.limit(page_size)
)
return db.execute(stmt).scalars().all()
total_devengado = func.sum(Daily.devengado).label("total_devengado")
top_entities = (
select(
Daily.entidad.label("entidad"),
total_devengado,
)
.group_by(Daily.entidad)
.order_by(total_devengado.desc(), Daily.entidad)
.limit(10)
.subquery()
)
ranked = (
select(
Daily.entidad.label("entidad"),
Daily.devengado.label("devengado"),
Daily.entidad_desc_area.label("entidad_desc_area"),
Daily.entidad_desc_entidad.label("entidad_desc_entidad"),
Daily.desc_programa.label("desc_programa"),
Daily.desc_actividad_proyecto.label("desc_actividad_proyecto"),
Daily.id.label("id"),
top_entities.c.total_devengado,
func.row_number().over(
partition_by=Daily.entidad,
order_by=(Daily.devengado.desc(), Daily.id),
).label("row_num"),
)
.join(top_entities, top_entities.c.entidad == Daily.entidad)
.subquery()
)
stmt = (
select(
ranked.c.entidad,
ranked.c.devengado,
ranked.c.entidad_desc_area,
ranked.c.entidad_desc_entidad,
ranked.c.desc_programa,
ranked.c.desc_actividad_proyecto,
)
.where(ranked.c.row_num <= page_size)
.order_by(
ranked.c.total_devengado.desc(),
ranked.c.entidad,
ranked.c.devengado.desc(),
ranked.c.id,
)
)
return db.execute(stmt).mappings().all()
@router.get('/timeseries', response_model=List[schemas.daily.TimeSeries])
@not_found
def fetch_timeseries(
db: Session = Depends(model.get_db),
):
return db.query(models.daily.TimeSeries).all()