-
Notifications
You must be signed in to change notification settings - Fork 0
/
manage.py
executable file
·194 lines (154 loc) · 4.48 KB
/
manage.py
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
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
import json
import os
COV = None
if os.environ.get('FLASK_COVERAGE'):
import coverage
COV = coverage.coverage(branch=True, include="app/*")
COV.start()
import datetime
from app.models import User, Role, Company, Indicators, Exchange
from app import create_app, db
# from populate_indicators import get_ratio_data
# from populate_companies import get_company_details
# from populate import get_company_sics, get_sectors_and_industries
# from app.external.companies import get_sic_code, get_sector_and_industry
app = create_app(os.getenv('FLASK_CONFIG') or 'default')
from flask_script import Manager, Shell
from flask_migrate import Migrate, MigrateCommand
migrate = Migrate(app, db)
manager = Manager(app)
# configure shell and migration commands
def make_shell_context():
return dict(app=app, db=db, User=User, Role=Role, Company=Company, Indicators=Indicators, Exchange=Exchange)
manager.add_command("shell", Shell(make_context=make_shell_context))
manager.add_command("db", MigrateCommand)
@manager.command
def test(coverage=False, pattern="test", quiet=True):
"""Runs the unit tests without coverage."""
if coverage and not os.environ.get('FLASK_COVERAGE'):
import sys
os.environ['FLASK_COVERAGE'] = '1'
os.execvp(sys.executable, [sys.executable] + sys.argv)
import unittest
loader = unittest.TestLoader()
loader.testMethodPrefix = pattern
tests = loader.discover('tests')
result = unittest.TextTestRunner(verbosity=2, buffer=quiet).run(tests)
if COV:
COV.stop()
COV.save()
print('Coverage summary')
COV.report()
basedir = os.path.abspath(os.path.dirname(__file__))
covdir = os.path.join(basedir, 'tmp/coverage')
COV.html_report(directory=covdir)
COV.erase()
if result.wasSuccessful():
return 0
else:
return 1
@manager.command
def cov():
"""Runs the unit tests with coverage."""
cov = coverage.coverage(branch=True, include='project/*')
cov.start()
tests = unittest.TestLoader().discover('tests')
unittest.TextTestRunner(verbosity=2).run(tests)
cov.stop()
cov.save()
print('Coverage Summary:')
cov.report()
basedir = os.path.abspath(os.path.dirname(__file__))
covdir = os.path.join(basedir, 'tmp/coverage')
cov.html_report(directory=covdir)
print('HTML version: file://%s/index.html' % covdir)
cov.erase()
@manager.command
def create_db():
"""Creates the db tables."""
db.create_all()
@manager.command
def drop_db():
"""Drops the db tables."""
db.drop_all()
@manager.command
def create_admin():
"""Creates the admin user."""
admin = db.session.query(User).filter_by(email="root2").first()
if admin:
admin.active = True
admin.confirmed_at = datetime.datetime.utcnow()
db.session.add(admin)
else:
db.session.add(User(email="root2", password="password", active=True, confirmed_at=datetime.datetime.utcnow()))
db.session.commit()
@manager.command
def make_data():
"""
Generate some fake data
"""
Company.generate_fake(3000)
Indicators.generate_fake(500)
# @manager.command
# def get_ratios():
# """
# Pull financial ratios
# """
# get_ratio_data()
#
# @manager.command
# def get_companies(throttle=True, index="NYSE", count=-1):
# """
# Pull company data
# """
# get_company_details(throttle=throttle, exchange=index, count=int(count))
# @manager.command
# def get_sics(symbol=""):
# """
#
# Args:
# symbol: company ticker symbol
#
# Returns:
#
# """
# if symbol:
# print get_sic_code(symbol)
# else:
# get_company_sics()
# @manager.command
# def get_sectors(symbol=""):
# """
#
# Args:
# symbol: company ticker symbol
#
# Returns:
#
# """
# if symbol:
# print get_sector_and_industry(symbol)
# else:
# get_sectors_and_industries()
@manager.command
def create_fixtures(model):
"""
Args:
model: company|indicators
Returns: None
"""
if model != "company":
model = "indicators"
models = {"company": Company, "indicators": Indicators}
d = {model: []}
objs = models[model].query.all()
for i in objs:
d[model].append(i.to_json())
with open("app/fixtures/{}.json".format(model), "w") as f:
f.write(json.dumps(d))
@manager.command
def deploy():
from flask.ext.migrate import upgrade
upgrade()
if __name__ == '__main__':
manager.run()