Alexander/FitTrackee
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

API: statistics by week - #9

Browse Source
This commit is contained in:
Sam 2018-06-06 00:22:24 +02:00
parent 7ac1109e1e
commit 600018b74f
3 changed files with 239 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
mpwo_api/mpwo_api/__init__.py
View File

@ -31,12 +31,14 @@ def create_app():
from .activities.activities import activities_blueprint # noqa
from .activities.records import records_blueprint # noqa
from .activities.sports import sports_blueprint # noqa
from .activities.stats import stats_blueprint # noqa
app.register_blueprint(users_blueprint, url_prefix='/api')
app.register_blueprint(auth_blueprint, url_prefix='/api')
app.register_blueprint(activities_blueprint, url_prefix='/api')
app.register_blueprint(records_blueprint, url_prefix='/api')
app.register_blueprint(sports_blueprint, url_prefix='/api')
app.register_blueprint(stats_blueprint, url_prefix='/api')
if app.debug:
logging.getLogger('sqlalchemy').setLevel(logging.WARNING)

72
mpwo_api/mpwo_api/activities/stats.py Normal file
View File

@ -0,0 +1,72 @@
from datetime import datetime
from flask import Blueprint, jsonify, request
from mpwo_api import appLog
from ..users.models import User
from ..users.utils import authenticate
from .models import Activity, convert_timedelta_to_integer
stats_blueprint = Blueprint('stats', __name__)
@stats_blueprint.route('/stats/<int:user_id>/by_week', methods=['GET'])
@authenticate
def get_activities(auth_user_id, user_id):
"""Get activities statistics for a user"""
try:
user = User.query.filter_by(id=user_id).first()
if not user:
response_object = {
'status': 'fail',
'message': 'User does not exist.'
}
return jsonify(response_object), 404
params = request.args.copy()
date_from = params.get('from')
date_to = params.get('to')
activities_list = {}
activities = Activity.query.filter(
Activity.user_id == user_id,
Activity.activity_date >= datetime.strptime(date_from, '%Y-%m-%d')
if date_from else True,
Activity.activity_date <= datetime.strptime(date_to, '%Y-%m-%d')
if date_to else True,
).order_by(
Activity.activity_date.asc()
).all()
for activity in activities:
week = f'W{datetime.strftime(activity.activity_date, "%U")}' # noqa
sport = activity.sports.label
if week not in activities_list:
activities_list[week] = {}
if sport not in activities_list[week]:
activities_list[week][sport] = {
'nb_activities': 0,
'total_distance': 0.,
'total_duration': 0,
}
activities_list[week][sport]['nb_activities'] += 1
activities_list[week][sport]['total_distance'] += \
float(activity.distance)
activities_list[week][sport]['total_duration'] += \
convert_timedelta_to_integer(activity.duration)
response_object = {
'status': 'success',
'data': {
'statistics': activities_list
}
}
code = 200
except Exception as e:
appLog.error(e)
response_object = {
'status': 'error',
'message': 'Error. Please try again or contact the administrator.'
}
code = 500
return jsonify(response_object), code

165
mpwo_api/mpwo_api/tests/test_stats_api.py Normal file
View File

@ -0,0 +1,165 @@
import json
def test_get_stats_no_activities(app, user_1):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.get(
f'/api/stats/{user_1.id}/by_week',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 200
assert 'success' in data['status']
assert data['data']['statistics'] == {}
def test_get_stats_no_user(app, user_1):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.get(
f'/api/stats/1000/by_week',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 404
assert 'fail' in data['status']
assert 'User does not exist.' in data['message']
def test_get_stats_all_activities(
app, user_1, sport_1_cycling, sport_2_running,
activity_cycling_user_1, activity_running_user_1
):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.get(
f'/api/stats/{user_1.id}/by_week',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 200
assert 'success' in data['status']
assert data['data']['statistics']['W00'] == \
{
'Cycling':
{
'nb_activities': 1,
'total_distance': 10.0,
'total_duration': 1024
}
}
assert data['data']['statistics']['W13'] == \
{
'Running':
{
'nb_activities': 1,
'total_distance': 12.0,
'total_duration': 6000
}
}
def test_get_stats_all_activities_week_13(
app, user_1, sport_1_cycling, sport_2_running,
activity_cycling_user_1, activity_running_user_1
):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.get(
f'/api/stats/{user_1.id}/by_week?from=2018-04-01&to=2018-04-30',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 200
assert 'success' in data['status']
assert data['data']['statistics'] == \
{
'W13':
{
'Running':
{
'nb_activities': 1,
'total_distance': 12.0,
'total_duration': 6000
}
}
}
def test_get_stats_all_activities_error(
app, user_1, sport_1_cycling, sport_2_running,
activity_cycling_user_1, activity_running_user_1
):
client = app.test_client()
resp_login = client.post(
'/api/auth/login',
data=json.dumps(dict(
email='test@test.com',
password='12345678'
)),
content_type='application/json'
)
response = client.get(
f'/api/stats/{user_1.id}/by_week?from="2018-04-01&to=2018-04-30',
headers=dict(
Authorization='Bearer ' + json.loads(
resp_login.data.decode()
)['auth_token']
)
)
data = json.loads(response.data.decode())
assert response.status_code == 500
assert 'error' in data['status']
assert 'Error. Please try again or contact the administrator.' \
in data['message']