from typing import List, Optional, AsyncIterable
import random
import uuid

import faust
from kafka import KafkaProducer

app = faust.App("demo")
producer = KafkaProducer()

class CustomerRecord(faust.Record, serializer="json"):
    msg_id: uuid.UUID
    customer: str
    amount: int
    number_of_purchases: int
    note: Optional[str]

customer_topic = app.topic("customer_topic", value_type=CustomerRecord)

completed_customer_topic = app.topic("completed_customer_topic", value_type=CustomerRecord)

def ring_up_customer(customer):
    customer.number_of_purchases += 1
    purchase_price = random.randint(1, 100)
    customer.amount += purchase_price
    print(f"{customer.customer} spent {purchase_price}")

def customer_is_done(customer):
    return (customer.number_of_purchases > 100
        or customer.amount > 5000
        or random.randint(1, 100) == 1)

@app.agent(customer_topic, concurrency=20)
async def purchase_processor(customers: AsyncIterable[CustomerRecord]):
    """Process customer purchases."""
    async for customer in customers:
        ring_up_customer(customer)
        if customer_is_done(customer):
            print(f"{customer.customer} spent {customer.amount / customer.number_of_purchases} on average.")
            print(customer)
            producer.send("completed_customer_topic", customer.dumps())
        else:
            producer.send("customer_topic", customer.dumps())

cfb54649f73cd68011561653869155fafba4ef91
<mask token>


class CustomerRecord(faust.Record, serializer='json'):
    msg_id: uuid.UUID
    customer: str
    amount: int
    number_of_purchases: int
    note: Optional[str]


<mask token>


def customer_is_done(customer):
    return (customer.number_of_purchases > 100 or customer.amount > 5000 or
        random.randint(1, 100) == 1)


<mask token>

<mask token>


class CustomerRecord(faust.Record, serializer='json'):
    msg_id: uuid.UUID
    customer: str
    amount: int
    number_of_purchases: int
    note: Optional[str]


<mask token>


def ring_up_customer(customer):
    customer.number_of_purchases += 1
    purchase_price = random.randint(1, 100)
    customer.amount += purchase_price
    print(f'{customer.customer} spent {purchase_price}')


def customer_is_done(customer):
    return (customer.number_of_purchases > 100 or customer.amount > 5000 or
        random.randint(1, 100) == 1)


@app.agent(customer_topic, concurrency=20)
async def purchase_processor(customers: AsyncIterable[CustomerRecord]):
    """Process customer purchases."""
    async for customer in customers:
        ring_up_customer(customer)
        if customer_is_done(customer):
            print(
                f'{customer.customer} spent {customer.amount / customer.number_of_purchases} on average.'
                )
            print(customer)
            producer.send('completed_customer_topic', customer.dumps())
        else:
            producer.send('customer_topic', customer.dumps())

<mask token>
app = faust.App('demo')
producer = KafkaProducer()


class CustomerRecord(faust.Record, serializer='json'):
    msg_id: uuid.UUID
    customer: str
    amount: int
    number_of_purchases: int
    note: Optional[str]


customer_topic = app.topic('customer_topic', value_type=CustomerRecord)
completed_customer_topic = app.topic('completed_customer_topic', value_type
    =CustomerRecord)


def ring_up_customer(customer):
    customer.number_of_purchases += 1
    purchase_price = random.randint(1, 100)
    customer.amount += purchase_price
    print(f'{customer.customer} spent {purchase_price}')


def customer_is_done(customer):
    return (customer.number_of_purchases > 100 or customer.amount > 5000 or
        random.randint(1, 100) == 1)


@app.agent(customer_topic, concurrency=20)
async def purchase_processor(customers: AsyncIterable[CustomerRecord]):
    """Process customer purchases."""
    async for customer in customers:
        ring_up_customer(customer)
        if customer_is_done(customer):
            print(
                f'{customer.customer} spent {customer.amount / customer.number_of_purchases} on average.'
                )
            print(customer)
            producer.send('completed_customer_topic', customer.dumps())
        else:
            producer.send('customer_topic', customer.dumps())

from typing import List, Optional, AsyncIterable
import random
import uuid
import faust
from kafka import KafkaProducer
app = faust.App('demo')
producer = KafkaProducer()


class CustomerRecord(faust.Record, serializer='json'):
    msg_id: uuid.UUID
    customer: str
    amount: int
    number_of_purchases: int
    note: Optional[str]


customer_topic = app.topic('customer_topic', value_type=CustomerRecord)
completed_customer_topic = app.topic('completed_customer_topic', value_type
    =CustomerRecord)


def ring_up_customer(customer):
    customer.number_of_purchases += 1
    purchase_price = random.randint(1, 100)
    customer.amount += purchase_price
    print(f'{customer.customer} spent {purchase_price}')


def customer_is_done(customer):
    return (customer.number_of_purchases > 100 or customer.amount > 5000 or
        random.randint(1, 100) == 1)


@app.agent(customer_topic, concurrency=20)
async def purchase_processor(customers: AsyncIterable[CustomerRecord]):
    """Process customer purchases."""
    async for customer in customers:
        ring_up_customer(customer)
        if customer_is_done(customer):
            print(
                f'{customer.customer} spent {customer.amount / customer.number_of_purchases} on average.'
                )
            print(customer)
            producer.send('completed_customer_topic', customer.dumps())
        else:
            producer.send('customer_topic', customer.dumps())

from typing import List, Optional, AsyncIterable
import random
import uuid

import faust
from kafka import KafkaProducer

app = faust.App("demo")
producer = KafkaProducer()

class CustomerRecord(faust.Record, serializer="json"):
    msg_id: uuid.UUID
    customer: str
    amount: int
    number_of_purchases: int
    note: Optional[str]

customer_topic = app.topic("customer_topic", value_type=CustomerRecord)

completed_customer_topic = app.topic("completed_customer_topic", value_type=CustomerRecord)

def ring_up_customer(customer):
    customer.number_of_purchases += 1
    purchase_price = random.randint(1, 100)
    customer.amount += purchase_price
    print(f"{customer.customer} spent {purchase_price}")

def customer_is_done(customer):
    return (customer.number_of_purchases > 100
        or customer.amount > 5000
        or random.randint(1, 100) == 1)

@app.agent(customer_topic, concurrency=20)
async def purchase_processor(customers: AsyncIterable[CustomerRecord]):
    """Process customer purchases."""
    async for customer in customers:
        ring_up_customer(customer)
        if customer_is_done(customer):
            print(f"{customer.customer} spent {customer.amount / customer.number_of_purchases} on average.")
            print(customer)
            producer.send("completed_customer_topic", customer.dumps())
        else:
            producer.send("customer_topic", customer.dumps())
