project.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
95
96
97
98
99
100
101
import pydantic_core
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_, select, func, desc
from presupuestov1 import model, schema, constant, schemas, models, search
from presupuestov1.routers._decorators import not_found
router = APIRouter(
prefix='/programa_proyecto',
tags=['programa_proyecto'],
)
###############################################################################
# GET
###############################################################################
@router.get('/{project_id}', response_model=List[schemas.project.ProyectoResumen])
@not_found
def fetch_programa_proyecto(
project_id: str,
db: Session = Depends(model.get_db),
):
ProyectoResumenes = models.project.ProyectoResumenes
project_items = db.query(ProyectoResumenes).filter(
ProyectoResumenes.codigo == project_id
).all()
# TODO if devengado > mucho_dinero: cache
for _ in ['entidad', 'objeto', 'finfun', 'acteco']:
for __ in project_items:
obj_ = getattr(__, _)
setattr(__, _, pydantic_core.from_json(obj_))
return project_items
@router.get('/{project_id}/proyectos')
@not_found
def fetch_programa_proyecto_proyectos(
project_id: str,
db: Session = Depends(model.get_db),
query: schema.SearchQueryParams = Depends(),
):
page = query.page or 1
page_size = 20
ProyectoResumenes = models.project.ProyectoResumenes
return db.execute(
select(
ProyectoResumenes.nivel,
ProyectoResumenes.codigo,
ProyectoResumenes.desc,
func.sum(ProyectoResumenes.devengado).label('devengado'),
).where(
ProyectoResumenes.codigo_padre == project_id,
ProyectoResumenes.nivel != 'programa',
).group_by(
ProyectoResumenes.nivel,
ProyectoResumenes.codigo,
ProyectoResumenes.desc,
).order_by(desc('devengado')).offset(
(page - 1) * page_size
).limit(page_size)
).mappings().all()
@router.get('/{project_id}/objetos')
@not_found
def fetch_programa_proyecto_objetos(
project_id: int,
db: Session = Depends(model.get_db),
query: schema.QueryParams = Depends(),
):
return []
@router.get('/{project_id}/finfuns')
@not_found
def fetch_programa_proyecto_finfuns(
project_id: int,
db: Session = Depends(model.get_db),
query: schema.QueryParams = Depends(),
):
return []
@router.get('/{project_id}/actecos')
@not_found
def fetch_programa_proyecto_actecos(
project_id: int,
db: Session = Depends(model.get_db),
query: schema.QueryParams = Depends(),
):
return []