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
|
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
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', methods=['PUT'])
def globeop():
download_data(request.args.get('workdate', datetime.date.today(),
lambda s: datetime.datetime.
strptime(s, "%Y-%m-%d").date()))
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
|
