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
|
import psycopg2
from psycopg2.extras import DictCursor
import datetime
import redis
from intex.load_intex_collateral import intex_data
from .globeop import download_data
from .insert_tranche_quotes import insert_quotes
from . import app, dawn_engine
from flask import request, g
import json
def get_db():
db = getattr(g, "db", None)
if db is None:
db = g.db = psycopg2.connect(
database="ET", user="et_user", host="debian", cursor_factory=DictCursor
)
return db
def get_queue():
q = getattr(g, "queue", None)
if q is None:
q = g.queue = redis.Redis(unix_socket_path="/run/redis/redis.sock")
return q
@app.teardown_appcontext
def close_db(error):
"""Closes the database again at the end of the request."""
db = getattr(g, "db", None)
if db:
db.close()
@app.route("/insert_intex_data")
def intex():
db = get_db()
intex_data(db, request.args.get("workdate", str(datetime.date.today())))
return "", 204
@app.route("/globeop/<fund>", methods=["PUT"])
def globeop(fund):
download_data(
dawn_engine,
request.args.get(
"workdate", datetime.date.today(), datetime.date.fromisoformat
),
fund,
)
return "", 202
@app.route("/insert_quotes", methods=["PUT"])
def insert_tranches():
insert_quotes(**request.args)
return "", 202
@app.route("/", methods=["POST"])
def run_tasks():
workdate = str(datetime.date.today())
with get_queue().pipeline() as pipe:
for dealname, reinvflag in request.form.items():
if reinvflag == "TRUE":
app.logger.info(f"Processing {dealname} with reinvestment")
else:
app.logger.info(f"Processing {dealname} without reinvestment")
pipe.rpush(
"tasks",
json.dumps(
{"fun": "build_portfolios", "args": [workdate, dealname, reinvflag]}
),
)
pipe.execute()
return "", 200
|
