Backend Changes

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from datetime import datetime
from typing import Optional
 
from models import Journey, Scooter, Ticket
from request_models import OffboardRequest
 
from tasks import create_offboarding_scooter_task
from request_models import OffboardTaskUpdatesRequest
 
app = FastAPI()
 
@app.post("/offboard-scooter")
async def offboard_scooter(request: OffboardRequest):
try: # Get journey and validate it belongs to the user and scooter
journey = await Journey.get(
id=request.journey_id,
user_id=request.user_id,
scooter_id=request.scooter_id,
status=Journey.STARTED
)
 
        if not journey:
            raise HTTPException(status_code=404, detail="Journey not found")
 
        # Update journey status
        journey.status = Journey.COMPLETED_REVIEWING
        journey.end_time = datetime.now()
        await journey.save()
 
        # Mark scooter as available
        scooter = journey.scooter
        scooter.is_available = True
        await scooter.save()
 
        # Create a task with offboarding photo
        ## Look task.py for details
        await create_offboarding_scooter_task(
            journey_id=request.journey_id,
            user_id=request.user_id,
            scooter_id=request.scooter_id,
            offboarding_photo_url=request.offboarding_photo_url
        )
 
        return {
            'status': 'success',
            'message': 'Scooter offboarded application got successfully',
            'journey_id': request.journey_id
        }
 
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
 
# your webhook for task updates
 
@app.post("/offboard-scooter-task-updates")
async def offboard_scooter_task_updates(request: OffboardTaskUpdatesRequest):
if(request.event == "TASK_COMPLETED"):
task = request.properties.task # Get the journey from the database
journey = await Journey.get(task.input_data.journey_id)
journey.status = Journey.COMPLETED
await journey.save()
 
        # Check if the photo is good
        if(task.output_data.result != "good"):
            # create a ticket to the user
            ticket = Ticket(
                user_id=user.id,
                journey_id=journey.id,
                amount=100
            )
            await ticket.insert()
 
            # update the user balance
            user = await User.get(task.input_data.user_id)
            user.balance -= 100
            await user.save()
 
    return {"status": "success"}
 

import httpx
 
api_key = "YOUR_API_KEY"
headers = {
    "Content-Type": "application/json",
    "Accept": "application/json",
    "Authorization": f"Bearer {api_key}"
}
 
my_webhook_url = "https://api.bricky.startup/offboard-scooter-task-updates"
 
async def create_offboarding_scooter_task(journey_id: str, user_id: str, scooter_id: str, offboarding_photo_url: str):
    queue_name = "scooter_offboarding_validation"
    url = f"http://api.qrambo.io/queues/{queue_name}/tasks"
 
    async with httpx.AsyncClient() as client:
        response = await client.post(
            url,
            json={
                "input_data": {
                    "journey_id": journey_id,
                    "user_id": user_id,
                    "scooter_id": scooter_id,
                    "offboarding_photo_url": offboarding_photo_url
                },
                "notification_settings": {
                    "strategy": "UNION",
                    "endpoint":my_webhook_url,
                    "headers": {},
                    "events_to_notify": ["TASK_COMPLETED"]
                }
            },
            headers=headers
        )
        return response.json()

from pydantic import BaseModel, Field
 
class OffboardRequest(BaseModel):
user_id: str
scooter_id: str
journey_id: str
offboarding_photo_url: str
 
class QueueProperties(BaseModel):
name: str
 
class TaskInputData(BaseModel):
journey_id: str = Field(alias="journey-id")
user_id: str = Field(alias="user-id")
scooter_id: str = Field(alias="scooter-id")
offboarding_photo_url: str = Field(alias="offboarding-photo-url")
 
class TaskOutputData(BaseModel):
result: str
 
class TaskProperties(BaseModel):
id: str
input_data: TaskInputData
output_data: TaskOutputData
 
class TaskEventProperties(BaseModel):
queue: QueueProperties
task: TaskProperties
 
class OffboardTaskUpdatesRequest(BaseModel):
event: str
properties: TaskEventProperties
 

from pydantic import BaseModel
from datetime import datetime
from typing import Optional
from beanie import Document, Indexed
 
class Scooter(Document):
    id: str = Indexed()
    is_available: bool = True
    last_location: Optional[str] = None
 
    class Settings:
        name = "scooters"
 
class Journey(Document):
    STARTED = 'started'
    COMPLETED = 'completed'
    COMPLETED_REVIEWING = 'completed_reviewing'
    COMPLETED_TICKET = 'completed_ticket'
 
    id: str = Indexed()
    user_id: str
    scooter_id: str
    status: str = STARTED
    start_time: datetime = datetime.now()
    end_time: Optional[datetime] = None
 
    class Settings:
        name = "journeys"
 
class Ticket(Document):
    id: str = Indexed()
    user_id: str
    journey_id: str
    amount: float
    created_at: datetime = datetime.now()
 
    class Settings:
        name = "tickets"