Firebase backend

.travis.yml

@@ -1,5 +1,19 @@
 matrix:
   include:
+    - language: python
+      dist: xenial
+      python:
+        - "3.7"
+      install:
+        - cd ./firebase/backend/
+        - pip install -r requirements.txt
+        - pip install codecov coverage
+        - cd -
+      # commands to run tests
+      script:
+        - cd ./firebase/backend/
+        - coverage run --omit="*/usr/*","*/.local/*","test_*" -m unittest discover
+        - codecov
     - language: python
       dist: xenial
       python:

firebase/backend/firebase_crud.py

@@ -0,0 +1,64 @@
+import itertools
+from google.cloud import firestore
+from multiprocessing.dummy import Pool as ThreadPool
+from multiprocessing import cpu_count
+from typing import Iterator
+
+from firebase_documents import FirebaseDocument
+
+# make sure you have GOOGLE_APPLICATION_CREDENTIALS environment variable pointing to service account credentials file
+# export GOOGLE_APPLICATION_CREDENTIALS="path/to/file"
+
+
+class BaseCrud:
+    NUMBER_OF_ITEMS_IN_SEQUENTIAL_BATCH = 500
+    NUMBER_OF_ITEMS_IN_PARALLEL_BATCH = 200
+
+    def __init__(self, connection: firestore.Client, collection_name: str):
+        self.connection = connection
+        self.collection_name = collection_name
+
+    def _create_document_ref(self, doc_id=None):
+        return self.connection.collection(self.collection_name).document(document_id=doc_id)
+
+    def create(self, content: FirebaseDocument, doc_id=None):
+        doc_id = doc_id if doc_id else content.get_firebase_id()
+
+        self._create_document_ref(doc_id=doc_id).create(content.to_firebase_dict())
+
+    def read(self, doc_id):
+        return self._create_document_ref(doc_id).get()
+
+    def create_multi(self, contents: Iterator[FirebaseDocument], parallel=False):
+        if parallel:
+            self._create_multi_parallel(contents, num_threads=cpu_count())
+        else:
+            self._create_multi_sequential(contents)
+
+    def _batch_commit(self, chunk: Iterator[FirebaseDocument]):
+        batch = self.connection.batch()
+        coll = self.connection.collection(self.collection_name)
+
+        for itm in chunk:
+            batch.create(coll.document(itm.get_firebase_id()), itm.to_firebase_dict())
+        batch.commit()
+
+    def _create_multi_sequential(self, contents: Iterator[FirebaseDocument]):
+        for chunk_of_rides in grouper(self.NUMBER_OF_ITEMS_IN_SEQUENTIAL_BATCH, contents):
+            self._batch_commit(chunk_of_rides)
+
+    # 187 sec for creating 12,000 documents with size of a SiriRide doc
+    def _create_multi_parallel(self, contents: Iterator[FirebaseDocument], num_threads):
+        pool = ThreadPool(num_threads)
+        pool.map(self._batch_commit, grouper(self.NUMBER_OF_ITEMS_IN_PARALLEL_BATCH, contents))
+        pool.close()
+        pool.join()
+
+
+def grouper(n, iterable: Iterator):
+    it = iter(iterable)
+    while True:
+        chunk = tuple(itertools.islice(it, n))
+        if not chunk:
+            return
+        yield chunk

firebase/backend/firebase_documents.py

@@ -0,0 +1,77 @@
+from datetime import datetime
+from google.cloud.firestore import GeoPoint
+from typing import Tuple, List
+
+
+class FirebaseDocument(object):
+    def to_firebase_dict(self):
+        raise NotImplementedError()
+
+    # noinspection PyMethodMayBeStatic
+    def get_firebase_id(self) -> object:
+        return None
+
+
+class Point(FirebaseDocument):
+    def __init__(self, lat: float, lon: float):
+        self.lat = lat
+        self.lon = lon
+
+    def to_firebase_dict(self):
+        return GeoPoint(self.lat, self.lon)
+
+
+class SiriPoint(FirebaseDocument):
+    def __init__(self, point: Point, rec_dt: datetime, pred_dt: datetime):
+        """
+        :param point: The geo location of the bus
+        :param rec_dt: The time that the bus has been at the point
+        :param pred_dt: The time siri client ask about bus location
+        """
+        self.rec_dt = rec_dt
+        self.pred_dt = pred_dt
+        self.point = point
+
+    @staticmethod
+    def from_str(line: str):
+        rec_dt, lat, lon, pred_dt = line.split('|')
+        rec_dt = datetime.strptime(rec_dt, '%Y-%m-%dT%H:%M:%S')
+        pred_dt = datetime.strptime(pred_dt, '%Y-%m-%dT%H:%M:%S')
+        point = Point(float(lat), float(lon))
+
+        return SiriPoint(point, rec_dt, pred_dt)
+
+    def to_firebase_dict(self):
+        return dict(rec_dt=self.rec_dt, pred_dt=self.pred_dt, point=self.point.to_firebase_dict())
+
+
+class SiriRide(FirebaseDocument):
+    def __init__(self, route_id: int, agency_id: int, bus_id: str, planned_start_datetime: datetime,
+                 route_short_name: str, stop_point_ref: Tuple[float, float], trip_id_to_date: int,
+                 points: List[SiriPoint], **kwargs):
+        """
+        :param route_id: Identifies a route
+        :param agency_id: Identifies a bus agency
+        :param bus_id: Identifies vehicle
+        :param planned_start_datetime: Departure time from a first stop
+        :param route_short_name: Short name of a route. This will often be a short, abstract identifier like "32"
+        :param stop_point_ref:
+        :param trip_id_to_date:
+        :param points: Realtime locations of vehicle
+        :param kwargs: Other metadata
+        """
+        self.route_id = route_id
+        self.agency_id = agency_id
+        self.bus_id = bus_id
+        self.planned_start_datetime = planned_start_datetime
+        self.route_short_name = route_short_name
+        self.stop_point_ref = stop_point_ref
+        self.trip_id_to_date = trip_id_to_date
+        self.points = points
+        self.metadata = kwargs
+
+    def to_firebase_dict(self):
+        return dict(route_id=self.route_id, agency_id=self.agency_id, bus_id=self.bus_id,
+                    planned_start_datetime=self.planned_start_datetime, route_short_name=self.route_short_name,
+                    stop_point_ref=self.stop_point_ref, trip_id_to_date=self.trip_id_to_date,
+                    points=[i.to_firebase_dict() for i in self.points], metadata=self.metadata)

firebase/backend/readme.md

@@ -0,0 +1,55 @@
+## How To
+
+#### Create CRUD Object
+For "Create \ Read \ ~~Update~~ \ ~~Delete~~" actions you should first create crud object: 
+```python
+from google.cloud import firestore
+from firebase_crud import BaseCrud
+
+conn = firestore.Client(project='openbus')
+siri_rides_crud = BaseCrud(connection=conn, collection_name='siri_rides')
+```
+
+#### Add New Document to DB
+```python
+from google.cloud import firestore
+from firebase_crud import BaseCrud
+from firebase_documents import Point
+
+
+conn = firestore.Client(project='openbus')
+siri_rides_crud = BaseCrud(connection=conn, collection_name='siri_rides')
+doc = Point(1.1, 2.2)
+siri_rides_crud.create(content=doc)
+```
+
+#### Get Document from DB
+```python
+from google.cloud import firestore
+from firebase_crud import BaseCrud
+from firebase_documents import Point
+
+conn = firestore.Client(project='openbus')
+siri_rides_crud = BaseCrud(connection=conn, collection_name='siri_rides')
+doc = Point(1.1, 2.2)
+siri_rides_crud.create(content=doc, doc_id='baz')
+result = siri_rides_crud.read(doc_id='baz')
+```
+
+#### Create SiriRide Document and Add it to DB
+```python
+from datetime import datetime
+from google.cloud import firestore
+from firebase_crud import BaseCrud
+from firebase_documents import SiriRide, SiriPoint, Point
+
+conn = firestore.Client(project='openbus')
+siri_rides_crud = BaseCrud(connection=conn, collection_name='siri_rides')
+doc = SiriRide(route_id=1001, agency_id=5, bus_id='23456', planned_start_datetime=datetime.fromtimestamp(1),
+               route_short_name='None', stop_point_ref=(1.1, 2.2), trip_id_to_date=57434, 
+               points=[SiriPoint(point=Point(1.1, 2.2), rec_dt=datetime.fromtimestamp(1),
+               pred_dt=datetime.fromtimestamp(1))], foo=4)
+
+siri_rides_crud.create(content=doc)
+
+```

firebase/backend/requirements.txt

@@ -0,0 +1,2 @@
+google-cloud-firestore
+mock-firestore