caronashow-api 0.7.1

Python client for CaronaShow APIs

caronashow-api

Python client for CaronaShow APIs.

Installation

pip install caronashow-api

Usage

The code below is a minimal version and not comprehensive of all APIs.

To view the API definitions and types, consult the .protobuf files or generated code.

Pooler API client

import asyncio

from caronashow.pooler.v1.pooler_connect import PoolerServiceClient
from caronashow.pooler.v1.pooler_pb2 import (
    MATCHING_ALGORITHM_RANDOM,
    ExecuteMatchingAlgorithmOnDatasetRequest,
)


async def main():
    # Create a client
    client = PoolerServiceClient("http://localhost:8080")

    # Make a request
    request = ExecuteMatchingAlgorithmOnDatasetRequest(
        dataset_id="dataset-123",
        algorithm=MATCHING_ALGORITHM_RANDOM,
    )

    # Call the service
    response = await client.execute_matching_algorithm_on_dataset(request)
    print(response.operation)


if __name__ == "__main__":
    asyncio.run(main())

Pooler API server

from caronashow.pooler.v1.pooler_connect import (
    PoolerService,
    PoolerServiceASGIApplication,
)
from caronashow.pooler.v1.pooler_pb2 import (
    ExecuteMatchingAlgorithmOnDatasetResponse,
    Operation,
    OperationState,
)


class Pooler(PoolerService):
    async def execute_matching_algorithm_on_dataset(self, request, ctx):
        print("Request headers:", ctx.request_headers())
        response = ExecuteMatchingAlgorithmOnDatasetResponse(
            operation=Operation(
                operation_id="op-123",
                dataset_id=request.dataset_id,
                state=OperationState.OPERATION_STATE_SUCCEEDED,
            )
        )
        return response


app = PoolerServiceASGIApplication(Pooler())

Onboarding API client

import asyncio

from caronashow.service.v1.onboarding_connect import OnboardingServiceClient
from caronashow.service.v1.onboarding_pb2 import (
    CompleteOnboardingRequest,
    DriverOfferSetup,
    GetOnboardingStatusRequest,
)
from caronashow.types.v1.common_pb2 import TimeOfDay
from caronashow.types.v1.user_pb2 import PARTICIPATION_MODE_OFFER


async def main():
    client = OnboardingServiceClient("http://localhost:8080")

    status = await client.get_onboarding_status(GetOnboardingStatusRequest())
    if not status.onboarding_completed:
        response = await client.complete_onboarding(
            CompleteOnboardingRequest(
                enrolled_in_vulnerability_filter=False,
                terms_accepted=True,
                participation_mode=PARTICIPATION_MODE_OFFER,
                offer=DriverOfferSetup(
                    vehicle_model="Honda Fit 2019",
                    vehicle_color="Prata",
                    vehicle_plate="ABC-1234",
                    offered_seats=4,
                    origin_address="Rua Exemplo, 100",
                    departure_time=TimeOfDay(hours=7, minutes=30),
                    return_time=TimeOfDay(hours=18, minutes=0),
                    accepting_requests=True,
                ),
            )
        )


if __name__ == "__main__":
    asyncio.run(main())

Types

from caronashow.types.v1 import user_pb2, common_pb2

# Create user objects
user = user_pb2.User(...)

# Common types
time = common_pb2.TimeOfDay(hours=7, minutes=30)
chat_pref = common_pb2.CHAT_PREFERENCE_OPEN

License

MIT License - see LICENSE file for details.