Letter Tracking System using a Postal Service API - with fully automa…

…ted testing

.gitignore

@@ -0,0 +1,28 @@
+.DS_Store
+.env
+.flaskenv
+*.pyc
+*.pyo
+env/
+env*
+dist/
+build/
+*.egg
+*.egg-info/
+_mailinglist
+.tox/
+.cache/
+.pytest_cache/
+.idea/
+docs/_build/
+.vscode
+
+# Coverage reports
+htmlcov/
+.coverage
+.coverage.*
+*,cover
+
+# Local testing
+local_testing/
+!local_testing/.gitkeep

Pipfile

@@ -0,0 +1,22 @@
+[[source]]
+name = "pypi"
+url = "https://pypi.org/simple"
+verify_ssl = true
+
+[dev-packages]
+pytest = "*"
+pytest-httpserver = "*"
+flask_alembic = "*"
+pytest-asyncio = "*"
+
+[packages]
+flask = "*"
+flask-sqlalchemy = "*"
+flask-cors = "*"
+python-dateutil = "*"
+requests = ">=2.27.1"
+SQLAlchemy = ">=1.3.12"
+flask_migrate = "*"
+
+[requires]
+python_version = "3.7"

README.md

@@ -0,0 +1,30 @@
+# Letter Tracking System using a Postal Service API
+
+### By Nicolas Poyiadjis
+https://www.linkedin.com/in/nicpoyia/
+
+## How to run and test
+
+### Environment Setup
+- Install python 3.7 & pipenv
+- `pipenv install` to install project's requirements
+- Before running the API, export the environment variable called LA_POSTE_API_KEY, which is the authorization key for La Poste API `export LA_POSTE_API_KEY=LA_POSTE_API_KEY_HERE`
+- Working tracking IDs are already stored in the sample SQLite database, therefore by retrieving statuses of all letters should return results. 
+- Execute command `flask run` to run the application's API
+- You can use postman_demo.json for a demo of the API
+
+### Automated Tests
+- Execute command `pipenv install -d` to install all dependencies required for development environment
+- Execute command `python -m pytest tests/` to run all automated tests
+
+There have been implemented 3 types of automated tests:
+- Unit tests, in which the model and data classes are being tested independently (the smallest units - the lowest layer)
+- Module tests (or service tests, or integration tests), in which the service containing the logic is tested by integrating a mock implementation of each external dependency, which in this case is La Poste API
+- End-to-end tests, in which the application is tested as a whole by running its HTTP API using mock dependencies
+
+In order to run the tests independently of production infrastructure, an independent SQLite database is generated on demand for testing purposes, i.e. before running the tests it is automatically created (if not yet) and the schema is initialized according to the application's migrations
+
+### Database Setup
+The database is already initialized with the updated schema and sample data to allow observing the application in action
+
+The command `flask db upgrade` is used to initialize the configured database with the appropriate schema (no data will be lost if it is already initialized).

app/__init__.py

@@ -0,0 +1,19 @@
+import os
+
+from flask import Flask
+from flask_cors import CORS
+from flask_migrate import Migrate
+from flask_sqlalchemy import SQLAlchemy
+
+from app.config import config
+
+app = Flask(__name__)
+CORS(app, origins="*", supports_credentials=True)
+config_name = os.getenv("FLASK_CONFIG") or "default"
+app.config.from_object(config[config_name])
+
+db = SQLAlchemy(app)
+Migrate(app, db)
+
+from .models import *
+from .views import *

app/config.py

@@ -0,0 +1,26 @@
+import os
+
+
+class Config:
+    SQLALCHEMY_DATABASE_URI = "sqlite:///la_poste_nicpoyia.db"
+    SQLALCHEMY_TRACK_MODIFICATIONS = False
+
+
+class DevelopmentConfig(Config):
+    ENV_TYPE = "development"
+    LA_POSTE_API_BASE_URL = "https://api.laposte.fr/ssu/v1"
+    LA_POSTE_API_KEY = os.environ.get('LA_POSTE_API_KEY')
+    APP_DEBUG = True
+
+
+class ProductionConfig(Config):
+    ENV_TYPE = "production"
+    LA_POSTE_API_BASE_URL = "https://api.laposte.fr/ssu/v1"
+    LA_POSTE_API_KEY = os.environ.get('LA_POSTE_API_KEY')
+
+
+config = {
+    "development": DevelopmentConfig,
+    "production": ProductionConfig,
+    "default": DevelopmentConfig,
+}

app/models/__init__.py

@@ -0,0 +1 @@
+__all__ = ["letter", "status_update"]

app/models/letter.py

@@ -0,0 +1,31 @@
+from sqlalchemy.sql import func
+
+from app import db
+
+
+class Letter(db.Model):
+    __tablename__ = "letter"
+
+    id = db.Column(db.Integer, primary_key=True)
+    tracking_number = db.Column(db.String(256), unique=True, index=True)
+    status = db.Column(db.String(191))
+    final = db.Column(db.Boolean(), index=True, default=False)
+    updated = db.Column(db.DateTime(timezone=True),
+                        server_default=func.now(),
+                        onupdate=func.now(),
+                        index=True)
+
+    def get_tracking_number(self) -> bool:
+        return self.tracking_number
+
+    def get_status_text(self) -> str:
+        return self.status
+
+    def is_final(self) -> bool:
+        return self.final
+
+    def get_last_update_timestamp(self) -> bool:
+        return self.updated
+
+    def make_final(self) -> None:
+        self.final = True

app/models/status_update.py

@@ -0,0 +1,22 @@
+from sqlalchemy import ForeignKey
+from sqlalchemy.sql import func
+
+from app import db
+
+
+class StatusUpdate(db.Model):
+    __tablename__ = "status_history"
+
+    id = db.Column(db.Integer, primary_key=True)
+    letter_id = db.Column(db.Integer, ForeignKey('letter.id'), index=True)
+    status = db.Column(db.String(191))
+    timestamp_tracked = db.Column(db.DateTime(timezone=True), server_default=func.now())
+
+    def get_letter_id(self) -> str:
+        return self.letter_id
+
+    def get_status_text(self) -> str:
+        return self.status
+
+    def get_tracking_timestamp(self) -> bool:
+        return self.timestamp_tracked

app/tracking_service/__init__.py

@@ -0,0 +1 @@
+__all__ = ["tracking_service", "tracking_response_dto", "tracking_exception"]

app/tracking_service/tracking_exception.py

@@ -0,0 +1,17 @@
+class CannotTrackLetterException(Exception):
+    def __init__(self, log_message: str):
+        self.log_message = log_message
+
+
+class CannotUpdateLetterTrackingException(Exception):
+    def __init__(self, log_message: str):
+        self.log_message = log_message
+
+
+class InvalidTrackingResponseException(Exception):
+    def __init__(self, invalid_object: str):
+        self.invalid_object = invalid_object
+
+
+class NoTrackingEventException(Exception):
+    pass

app/tracking_service/tracking_response_dto.py

@@ -0,0 +1,75 @@
+from datetime import datetime
+from typing import List
+
+from .tracking_exception import InvalidTrackingResponseException, NoTrackingEventException
+
+
+class _TrackingEventDto:
+    date: datetime
+    label: str
+
+    def __init__(self, date: datetime, label: str) -> None:
+        super().__init__()
+        self.date = date
+        self.label = label
+
+    @staticmethod
+    def from_json_dict(json_dict):
+        """
+        Factory method which generates an object of the class by using the received raw JSON data
+        :param json_dict: Dictionary with JSON data
+        :return: _TrackingEventDto
+        :raises:
+             InvalidTrackingResponseException: In case of invalid payload received in the response
+        """
+        date = json_dict.get('date')
+        if not date:
+            raise InvalidTrackingResponseException("event.date")
+        label = json_dict.get('label')
+        if not label:
+            raise InvalidTrackingResponseException("event.label")
+        return _TrackingEventDto(date, label)
+
+
+class TrackingResponseDto:
+    # Whether the tracking is final, i.e. no further changes will apply
+    is_final: bool
+    # Tracking events in anti-chronological order
+    events: List[_TrackingEventDto]
+
+    def __init__(self, is_final: bool, events: List[_TrackingEventDto]) -> None:
+        super().__init__()
+        self.is_final = is_final
+        self.events = events
+        self.events.sort(key=lambda ev: ev.date, reverse=True)
+
+    @staticmethod
+    def from_json_dict(json_dict):
+        """
+        Factory method which generates an object of the class by using the received raw JSON data
+        :param json_dict: Dictionary with JSON data
+        :return: TrackingResponseDto
+        :raises:
+             InvalidTrackingResponseException: In case of invalid payload received in the response
+        """
+        shipment_obj = json_dict.get('shipment')
+        if not shipment_obj:
+            raise InvalidTrackingResponseException("shipment")
+        if 'event' not in shipment_obj:
+            raise InvalidTrackingResponseException("event")
+        events_data_list = shipment_obj.get('event')
+        events_list = [_TrackingEventDto.from_json_dict(event_dict) for event_dict in events_data_list]
+        return TrackingResponseDto(
+            is_final=shipment_obj.get('isFinal'),
+            events=events_list
+        )
+
+    def get_last_event_status(self) -> str:
+        """
+        :return: Status label of last tracking event
+        :raises:
+            NoTrackingEventException If no event is available
+        """
+        if not self.events:
+            raise NoTrackingEventException()
+        return self.events[0].label