Middle

The screenshots contain Python code for an asynchronous function to create an order in a web service. The code involves: - Parsing JSON request data for user_id, items, and promo_code. - Calculating the total price of items. - Optionally applying a discount by calling an external discount service. - Inserting the order into a PostgreSQL database. - Reserving items by calling a warehouse service. - Publishing an order created event to a Kafka adapter. - Updating the order status to 'processed'. The task implied by the code is: "Implement an asynchronous API endpoint to create an order that: - Validates input data. - Calculates the total order price. - Applies a discount if a promo code is provided by calling an external discount service. - Inserts the order into a PostgreSQL database and retrieves the new order ID. - Calls a warehouse service to reserve the ordered items. - Publishes an order created event to a Kafka adapter. - Updates the order status to 'processed' in the database. - Handles potential errors such as discount service disconnection gracefully." No explicit question text is present, but the task is to implement or improve this order creation workflow with asynchronous calls, error handling, and database interaction.

sobes.tech AI

Answer from AI

To implement an asynchronous API endpoint for creating an order in Python, you can use asyncio and libraries that support asynchronous operations, such as aiohttp for the web server, asyncpg for PostgreSQL, and asynchronous HTTP clients for calling external services.

Main steps:

Input validation — check that user_id, items, and optionally promo_code are correct.
Calculate total cost — sum the prices of items from items.
Apply discount — if there is a promo_code, call the external discount service asynchronously, handle errors (e.g., timeouts, service unavailability), and adjust the final amount.
Insert order into DB — asynchronously insert a record into PostgreSQL, get order_id.
Reserve items — call the warehouse service asynchronously for reservation, handle errors.
Publish event — send a message to Kafka (via an asynchronous adapter).
Update order status — change status to "processed" in the database.

Example simplified code using aiohttp and asyncpg:

import aiohttp
import asyncpg
from aiohttp import web

async def create_order(request):
    data = await request.json()
    user_id = data.get('user_id')
    items = data.get('items')
    promo_code = data.get('promo_code')

    if not user_id or not items:
        return web.json_response({'error': 'Invalid input'}, status=400)

    total_price = sum(item['price'] * item['quantity'] for item in items)

    # Apply discount
    if promo_code:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(f'http://discount-service/apply?code={promo_code}&amount={total_price}') as resp:
                    if resp.status == 200:
                        discount_data = await resp.json()
                        total_price = discount_data.get('discounted_amount', total_price)
        except Exception:
            # Log error but continue without discount
            pass

    pool = request.app['db_pool']
    async with pool.acquire() as conn:
        async with conn.transaction():
            order_id = await conn.fetchval(
                'INSERT INTO orders(user_id, total_price, status) VALUES($1, $2, $3) RETURNING id',
                user_id, total_price, 'pending'
            )

            # Reserve items
            try:
                async with aiohttp.ClientSession() as session:
                    reserve_payload = {'order_id': order_id, 'items': items}
                    async with session.post('http://warehouse-service/reserve', json=reserve_payload) as resp:
                        if resp.status != 200:
                            raise Exception('Warehouse reservation failed')
            except Exception:
                await conn.execute('UPDATE orders SET status=$1 WHERE id=$2', 'failed', order_id)
                return web.json_response({'error': 'Failed to reserve items'}, status=500)

            # Publish event to Kafka (pseudo-code)
            # await kafka_producer.send('order_created', {'order_id': order_id, 'user_id': user_id})

            await conn.execute('UPDATE orders SET status=$1 WHERE id=$2', 'processed', order_id)

    return web.json_response({'order_id': order_id, 'total_price': total_price})

Important points:

All external calls and database operations are performed asynchronously.
Errors from external services are handled to prevent the entire process from failing.
The database transaction guarantees data integrity.
Publishing to Kafka can be implemented via an asynchronous client.

This approach ensures API responsiveness and resilience to external service failures.

Previous Next