ingestion of all data

server/src/app.py

@@ -94,30 +94,61 @@ async def sample_route(request):
     return json(sample_dataset)
 
 
-@app.route('/ingest', methods=["POST"])
+@app.route('/ingest', methods=["GET"])
 @compress.compress()
 async def ingest(request):
-    """Accept POST requests with a list of years to import.
-        Query parameter name is 'years', and parameter value is
-        a comma-separated list of years to import.
-        Ex. '/ingest?years=2015,2016,2017'
     """
+    Query parameters:
+        years:
+            a comma-separated list of years to import.
+            Ex. '/ingest?years=2015,2016,2017'
+        limit:
+            the max number of records per year
+        querySize:
+            the number of records per request to socrata
+
+    Counts:
+        These are the counts you can expect if you do the full ingest:
+
+        2015: 237305
+        2016: 952486
+        2017: 1131558
+        2018: 1210075
+        2019: 1308093
+        2020: 319628 (and counting)
+
+        GET https://data.lacity.org/resource/{ID}.json?$select=count(srnumber)
+
+    Hint:
+        Run /ingest without params to get all socrata data
+    """
+
+    # parse params
+    defaults = app.config['Settings']['Ingestion']
+
+    years = request.args.get('years', defaults['YEARS'])
+    limit = request.args.get('limit', defaults['LIMIT'])
+    querySize = request.args.get('querySize', defaults['QUERY_SIZE'])
+
+    # validate params
     current_year = datetime.now().year
-    querySize = request.args.get("querySize", None)
-    limit = request.args.get("limit", None)
-    ALLOWED_YEARS = [year for year in range(2015, current_year+1)]
-    if not request.args.get("years"):
-        return json({"error": "'years' parameter is required."})
-    years = set([int(year) for year in request.args.get("years").split(",")])
-    if not all(year in ALLOWED_YEARS for year in years):
-        return json({"error":
-                    f"'years' param values must be one of {ALLOWED_YEARS}"})
+    allowed_years = [year for year in range(2015, current_year+1)]
+    years = set([int(year) for year in years.split(',')])
+    if not all(year in allowed_years for year in years):
+        return json({
+            'error': f"'years' param values must be one of {allowed_years}"
+        })
+
+    limit = int(limit)
+    querySize = int(querySize)
+    querySize = min([limit, querySize])
+
+    # get data
     loader = DataHandler(app.config['Settings'])
-    loader.populateFullDatabase(yearRange=years,
-                                querySize=querySize,
-                                limit=limit)
-    return_data = {'response': 'ingest ok'}
-    return json(return_data)
+    data = await loader.populateDatabase(years=years,
+                                         limit=limit,
+                                         querySize=querySize)
+    return json(data)
 
 
 @app.route('/update')

server/src/services/databaseOrm.py

@@ -1,4 +1,4 @@
-from sqlalchemy import Column, Integer, String, DateTime, Float
+from sqlalchemy import Column, Integer, String, DateTime, Float, JSON
 from sqlalchemy.ext.declarative import declarative_base
 
 
@@ -16,20 +16,33 @@ def _asdict(self):
 
 class Ingest(Base, Mixin):
     __tablename__ = 'ingest_staging_table'
-    srnumber = Column(String(50), primary_key=True, unique=True)
+
+    # a temporary primary key
+    id = Column(Integer, primary_key=True, autoincrement=True)
+
+    # becomes the primary key after deduplication
+    srnumber = Column(String)
+
+    # dates
     createddate = Column(DateTime)
     updateddate = Column(DateTime)
+    servicedate = Column(DateTime)
+    closeddate = Column(DateTime)
+
+    # about
+    requesttype = Column(String)
+    requestsource = Column(String)
     actiontaken = Column(String)
     owner = Column(String)
-    requesttype = Column(String)
     status = Column(String)
-    requestsource = Column(String)
     createdbyuserorganization = Column(String)
     mobileos = Column(String)
     anonymous = Column(String)
     assignto = Column(String)
-    servicedate = Column(String)
-    closeddate = Column(String)
+
+    # location
+    latitude = Column(Float)
+    longitude = Column(Float)
     addressverified = Column(String)
     approximateaddress = Column(String)
     address = Column(String)
@@ -38,96 +51,17 @@ class Ingest(Base, Mixin):
     streetname = Column(String)
     suffix = Column(String)
     zipcode = Column(String)
-    latitude = Column(String)
-    longitude = Column(String)
-    location = Column(String)
-    tbmpage = Column(String)
-    tbmcolumn = Column(String)
-    tbmrow = Column(String)
+    location = Column(JSON)
+
+    # politics
     apc = Column(String)
-    cd = Column(String)
+    cd = Column(Integer)
     cdmember = Column(String)
-    nc = Column(String)
+    nc = Column(Integer)
     ncname = Column(String)
     policeprecinct = Column(String)
 
-
-insertFields = {'srnumber': String(50),
-                'createddate': DateTime,
-                'updateddate': DateTime,
-                'actiontaken': String(30),
-                'owner': String(10),
-                'requesttype': String(30),
-                'status': String(20),
-                'requestsource': String(30),
-                'createdbyuserorganization': String(16),
-                'mobileos': String(10),
-                'anonymous': String(10),
-                'assignto': String(20),
-                'servicedate': String(30),
-                'closeddate': String(30),
-                'addressverified': String(16),
-                'approximateaddress': String(20),
-                'address': String(250),
-                'housenumber': String(10),
-                'direction': String(10),
-                'streetname': String(50),
-                'suffix': String(10),
-                'zipcode': Integer,
-                'latitude': Float,
-                'longitude': Float,
-                'location': String(250),
-                'tbmpage': Integer,
-                'tbmcolumn': String(10),
-                'tbmrow': Float,
-                'apc': String(30),
-                'cd': Float,
-                'cdmember': String(30),
-                'nc': Float,
-                'ncname': String(100),
-                'policeprecinct': String(30)}
-
-
-readFields = {'SRNumber': str,
-              'CreatedDate': str,
-              'UpdatedDate': str,
-              'ActionTaken': str,
-              'Owner': str,
-              'RequestType': str,
-              'Status': str,
-              'RequestSource': str,
-              'MobileOS': str,
-              'Anonymous': str,
-              'AssignTo': str,
-              'ServiceDate': str,
-              'ClosedDate': str,
-              'AddressVerified': str,
-              'ApproximateAddress': str,
-              'Address': str,
-              'HouseNumber': str,
-              'Direction': str,
-              'StreetName': str,
-              'Suffix': str,
-              'ZipCode': str,
-              'Latitude': str,
-              'Longitude': str,
-              'Location': str,
-              'TBMPage': str,
-              'TBMColumn': str,
-              'TBMRow': str,
-              'APC': str,
-              'CD': str,
-              'CDMember': str,
-              'NC': str,
-              'NCName': str,
-              'PolicePrecinct': str}
-
-
-tableFields = ['srnumber', 'createddate', 'updateddate', 'actiontaken',
-               'owner', 'requesttype', 'status', 'requestsource',
-               'createdbyuserorganization', 'mobileos', 'anonymous',
-               'assignto', 'servicedate', 'closeddate', 'addressverified',
-               'approximateaddress', 'address', 'housenumber', 'direction',
-               'streetname', 'suffix', 'zipcode', 'latitude', 'longitude',
-               'location', 'tbmpage', 'tbmcolumn', 'tbmrow', 'apc', 'cd',
-               'cdmember', 'nc', 'ncname', 'policeprecinct']
+    # misc
+    tbmpage = Column(String)
+    tbmcolumn = Column(String)
+    tbmrow = Column(Integer)

server/src/services/socrataClient.py

@@ -0,0 +1,48 @@
+"""
+This is a simple wrapper for Socrata that handles a couple of issues:
+1. retrying requests in the event of a timeout or other failure
+2. grabbing the dataset id from the config based on the year
+3. automatically closing the client when we're done using it
+
+Usage is like this:
+    socrata = SocrataClient()
+    results = socrata.get(year, **kwargs)
+
+kwargs are all of the normal socrata kwargs - select, limit, etc.
+"""
+
+from sodapy import Socrata
+
+
+class SocrataClient:
+    def __init__(self, config=None):
+        config = config['Socrata']
+
+        domain = config['DOMAIN']
+        token = None if config['TOKEN'] == 'None' else config['TOKEN']
+        timeout = int(config['TIMEOUT'])
+
+        self.client = Socrata(domain, token, timeout=timeout)
+        self.attempts = int(config['ATTEMPTS'])
+        self.config = config
+
+    def __del__(self):
+        self.client.close()
+
+    def dataset_id(self, year):
+        return self.config['AP' + str(year)]
+
+    def get(self, year, **kwargs):
+        id = self.dataset_id(year)
+        for attempt in range(self.attempts):
+            try:
+                return self.client.get(id, **kwargs)
+            except Exception as e:
+                if attempt < self.attempts - 1:
+                    continue
+                else:
+                    raise e
+
+    def get_metadata(self, year):
+        id = self.dataset_id(year)
+        return self.client.get_metadata(id)