User data model

database.py

@@ -85,3 +85,16 @@ def db(self):
                     current_app.config['CLOUDANT_DB_NAME'],
                     throw_on_exists=False)
             return ctx.cloudant_db
+
+
+cloudant_ext = FlaskCloudant()
+
+
+class MetaDocument(type):
+
+    name = None
+    _db = None
+
+    def __getitem__(cls, key):
+        return cls.db[key]
+

models.py

@@ -0,0 +1,95 @@
+
+from database import cloudant_ext
+import cloudant
+
+
+class Document:
+    """Metaclass to represent mapped Cloudant objects."""
+
+    def __init__(self, _rev=None):
+        """Construct a document.
+
+        Sets the type attibute as the lowercase version of the
+        most derived class name.
+        """
+        self.type = type(self).__name__.lower()
+        self._rev = _rev
+
+    def save(self):
+        """Save the object to the Cloudant database, create if necessary."""
+        if self.exists():
+            doc = cloudant_ext.db[self._id]
+            doc.update(vars(self))
+            doc.save()
+        else:
+            doc = vars(self)
+            del doc['_rev']
+            cloudant_ext.db.create_document(doc)
+
+    @classmethod
+    def from_raw(cls, raw):
+        """Construct an object from the raw dict-like Cloudant document."""
+        del raw['type']
+        return cls(**raw)
+
+    @classmethod
+    def get(cls, _id):
+        """Get document corresponding to ID.
+
+        Args:
+            _id: primary key/ID of object to be retrieved
+        """
+        doc = cloudant_ext.db[_id]
+        return cls.from_raw(doc)
+
+    @classmethod
+    def contains(cls, _id):
+        """Return True if document with _id exists in DB.
+
+        Args:
+            _id: primary key/ID of object to be retrieved
+        """
+        return cloudant.document.Document(cloudant_ext.db, _id).exists()
+
+    def exists(self):
+        """Return True if this document exists in database."""
+        return User.contains(self._id)
+
+    @classmethod
+    def all(cls):
+        """Get all instances of specific type defined by derived class."""
+        selector = {'type': {'$eq': cls.__name__.lower()}}
+        users = cloudant_ext.db.get_query_result(selector)
+        return [cls.from_raw(user) for user in users]
+
+    @classmethod
+    def delete(cls, _id):
+        """Delete document with _id if exists.
+
+        Args:
+            _id: primary key/ID of object to be deleted
+        """
+        if cls.contains(_id):
+            cloudant_ext.db[_id].delete()
+
+
+class User(Document):
+    """Representation of a Segmund/Strava user."""
+
+    def __init__(self,
+                 _id,
+                 name,
+                 firstname,
+                 lastname,
+                 access_token,
+                 expires_at,
+                 refresh_token,
+                 **kwargs):
+        super().__init__(**kwargs)
+        self._id = _id
+        self.name = name
+        self.firstname = firstname
+        self.lastname = lastname
+        self.access_token = access_token
+        self.expires_at = expires_at
+        self.refresh_token = refresh_token

segmund.py

@@ -1,15 +1,13 @@
-from cloudant import Cloudant
-from cloudant.document import Document
-from flask import Flask, render_template, request, jsonify, redirect
-import atexit
+
 import os
 import json
-import requests
-import time
+
+from flask import Flask, render_template, request, jsonify, redirect
 
 import strava
 import date_utils
-import database
+from database import cloudant_ext
+import models
 
 app = Flask(__name__, static_url_path='')
 
@@ -33,7 +31,7 @@
     vcap_app = json.loads(os.getenv('VCAP_APPLICATION'))
     current_domain = "https://{}".format(vcap_app['application_uris'][0])
 
-cloudant_ext = database.FlaskCloudant(app)
+cloudant_ext.init_app(app)
 
 @app.route('/')
 def root():
@@ -53,9 +51,9 @@ def get_registration_result():
 def get_hop_segment_results():
     activity_date = request.args.get('date')
     if activity_date is None:
-        leader_results = strava_service.hop_alltime_leaders(cloudant_ext.db)
+        leader_results = strava_service.hop_alltime_leaders()
     else:
-        leader_results = strava_service.get_hop_activities(cloudant_ext.db, activity_date)
+        leader_results = strava_service.get_hop_activities(activity_date)
     # For now, provide a rolling window of 5 thursdays -- eventually this will just come from DB
     return render_template('results.html', results=leader_results, date=activity_date, dates=date_utils.thursdays(5))
 
@@ -64,7 +62,7 @@ def get_activities():
     activity_date = request.args.get('date')
     #TODO Validate date
     if cloudant_ext.client:
-        return jsonify(strava_service.get_hop_activities(cloudant_ext.db, activity_date))
+        return jsonify(strava_service.get_hop_activities(activity_date))
     else:
         print('No database')
         return jsonify([])
@@ -75,40 +73,31 @@ def get_activities():
 # */
 @app.route('/exchange_token', methods=['GET'])
 def register_user():
-    state = request.args.get('state')
     auth_code = request.args.get('code')
-    scope = request.args.get('scope')
 
     user = strava_service.register_user(auth_code)
 
+    # TODO: remove this condition?  I don't think the function can return None
     if user is None:
         return "Failed to register user with Strava"
 
-    if Document(cloudant_ext.db, user['_id']).exists():
-        print("User id={} exists already, Updating.".format(user['_id']))
-        user_document = cloudant_ext.db[user['_id']]
-        user_document.update(user)
-        user_document.save()
+    if user.exists():
+        print("User id={} exists already, Updating.".format(user._id))
     else:
-        print ("Creating User: {}".format(user))
-        user_document = cloudant_ext.db.create_document(user)
-        user['_id'] = user_document['_id']
+        print("Creating User: {}".format(user))
+        user.save()
 
-    if user_document.exists():
-        print('Doc with _id={}'.format(user['_id']))
+    if user.exists():
+        print('Doc with _id={}'.format(user._id))
 
-    return redirect('/users?firstname={}'.format(user['firstname']), code=302)
+    return redirect('/users?firstname={}'.format(user.firstname), code=302)
 
 @app.route('/users', methods=['GET'])
 def get_users():
     firstname = request.args.get('firstname')
-    selector = {'type': {'$eq': 'user'}}
-    docs = cloudant_ext.db.get_query_result(selector)
+    docs = [vars(u) for u in models.User.all()]
     return render_template('users.html', users=docs, firstname=firstname)
 
-@atexit.register
-def shutdown():
-    pass
 
 if __name__ == '__main__':
     app.run(host='0.0.0.0', port=port, debug=True)

strava.py

@@ -2,9 +2,12 @@
 A crappy Strava helper module
 """
 from typing import Dict
-import requests
+
 import stravalib
+
 import date_utils
+import models
+
 
 class Strava:
 
@@ -40,18 +43,14 @@ def register_user(self, auth_code):
 
         athlete = client.get_athlete()
 
-        user = {
-            "_id": str(athlete.id),
-            "type": "user",
-            "name": athlete.username,
-            "firstname": athlete.firstname,
-            "lastname": athlete.lastname,
-            "access_token": client.access_token,
-            "expires_at": client.token_expires_at,
-            "refresh_token": client.refresh_token
-        }
+        return models.User(_id=str(athlete.id),
+                           name=athlete.username,
+                           firstname=athlete.firstname,
+                           lastname=athlete.lastname,
+                           access_token=client.access_token,
+                           expires_at=client.token_expires_at,
+                           refresh_token=client.refresh_token)
 
-        return user
 
     # /* Given a segment, return club leaders for a given date range
     # *
@@ -66,16 +65,18 @@ def segment_leaderboard(self, segment_id, token, club_id, date_range):
                          for entry in leaderboard]
         return leader_result
 
-    # Gather hop segment leaders for a given date
-    def hop_alltime_leaders(self, db):
-
-        #TODO: Move this kind of stuff into DB service
-        john = db['326452']
-        johns_token = self.get_user_access_token(john, db)
+    def hop_alltime_leaders(self):
+        """Gather all-time hop segment leaders."""
+        # TODO: clarify if we really should use John for this.  Couldn't
+        # this use the current user's token?  Maybe that would be slightly
+        # safer for rate limits?
+        john = models.User.get('326452')
+        johns_token = self.get_user_access_token(john)
 
         segment_leaders = {}
-        for (segment_id,name) in self.hop_segments.items():
-            segment_leaders[name] = self.segment_leaderboard(segment_id, johns_token, self.hop_club, "this_month")
+        for (segment_id, name) in self.hop_segments.items():
+            segment_leaders[name] = self.segment_leaderboard(
+                segment_id, johns_token, self.hop_club, "this_month")
         return segment_leaders
 
     def get_leaderboard_entry_dict(self,
@@ -97,7 +98,7 @@ def get_leaderboard_entry_dict(self,
             "start_date_local": entry.start_date_local,
             "rank": entry.rank}
 
-    def get_hop_activities(self, db, hop_date):
+    def get_hop_activities(self, hop_date):
         """
             Compile activity segment results for HOP on a given date
 
@@ -110,24 +111,24 @@ def get_hop_activities(self, db, hop_date):
                 - Persist Activity Segment Data to DB
              - Build leaderboard from DB data
         """
-        selector = {'type': {'$eq': 'user'}}
-        users = db.get_query_result(selector)
+        users = models.User.all()
 
         segment_leaders = {segment: [] for segment in self.hop_segments.values()}
         start_date = hop_date
         end_date = date_utils.next_day(start_date)
 
         print("Searching for activities between start_date={} and end_date={}".format(start_date, end_date))
         for user in users:
-            activities = self.get_public_hop_activities(user, db, start_date, end_date)
+            activities = self.get_public_hop_activities(user, start_date, end_date)
             for activity in activities:
                 for effort in activity.segment_efforts:
                     if str(effort.segment.id) in self.hop_segments.keys():
                         segment_leaders.get(effort.name).append({
                             "segment_name":effort.name,
                             "rank": "N/A",
                             "athlete_id": str(activity.athlete.id),
-                            "athlete_name": "{}, {}".format(user['lastname'], user['firstname']),
+                            "athlete_name": 
+                                f"{user.firstname}, {user.lastname}",
                             "activity": activity.name,
                             "start_date_local": str(activity.start_date_local),
                             "elapsed_time": str(effort.elapsed_time),
@@ -136,37 +137,39 @@ def get_hop_activities(self, db, hop_date):
                             })
         return segment_leaders
 
-    def get_public_activities(self, user, db, start_date, end_date):
+    def get_public_activities(self, user, start_date, end_date):
         """Return detailed public activities for a given user beteen two dates"""
-        token = self.get_user_access_token(user, db)
+        token = self.get_user_access_token(user)
         client = stravalib.Client(token)
         activities = client.get_activities(after=start_date, before=end_date)
-        return map(lambda activity : client.get_activity(activity.id, True),
-                    filter(lambda activity: not activity.private, activities))
+        return map(lambda activity: client.get_activity(activity.id, True),
+                   filter(lambda activity: not activity.private, activities))
 
-    def get_public_hop_activities(self, user, db, start_date, end_date):
+    def get_public_hop_activities(self, user, start_date, end_date):
         """Return any activities that contain the hop segment"""
         return filter(lambda activity: self.has_hop_segment(activity),
-                self.get_public_activities(user, db, start_date, end_date))
+                      self.get_public_activities(user, start_date, end_date))
 
     def has_hop_segment(self, activity):
         """True if an activity contains the HOP segment"""
         return any(effort.segment.id == self.hop_segment_id
-                        for effort in activity.segment_efforts)
+                   for effort in activity.segment_efforts)
 
-    def get_user_access_token(self, user, db):
+    def get_user_access_token(self, user):
         """Return user access token, if expired refresh it and persis to DB"""
-        if date_utils.is_expired(user['expires_at']):
-            print("token is expired for user={}".format(user))
+        if date_utils.is_expired(user.expires_at):
+            print("token is expired for user={}".format(user._id))
             client = stravalib.Client()
-            refresh_response = client.refresh_access_token(client_id=self.client_id,
-                client_secret=self.secret, refresh_token=user['refresh_token'])
+            refresh_response = client.refresh_access_token(
+                client_id=self.client_id,
+                client_secret=self.secret,
+                refresh_token=user.refresh_token)
             # update the user with their new token and expiration
-            user_document = db[user['_id']]
-            user_document['access_token'] = refresh_response['access_token']
-            user_document['refresh_token'] = refresh_response['refresh_token']
-            user_document['expires_at'] = refresh_response['expires_at']
-            user_document.save()
+
+            user.access_token = refresh_response['access_token']
+            user.refresh_token = refresh_response['refresh_token']
+            user.expires_at = refresh_response['expires_at']
+            user.save()
             return refresh_response['access_token']
         else:
-            return user['access_token']
+            return user.access_token