highsmallxu

from typing import NamedTuple
import collections
from dataclasses import dataclass

# regular class
class TransactionObject:
    
    def __init__(self, sender, receiver, date, amount):
        self.sender = sender
        self.receiver = receiver

highsmallxu

highsmallxu

trans_obj = TransactionObject(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender='jojo')
trans_named_tuple = TransactionNamedTuple(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender='jojo')
trans_named_tuple_typing = TransactionNamedTupleTyping(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender='jojo')
trans_dataclass = TransactionDataClass(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender='jojo')
trans_dataclass_slot = TransactionDataClassWithSlot(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender='jojo')

highsmallxu

from timeit import timeit
create_obj = timeit('TransactionObject(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
create_named_tuple = timeit('TransactionNamedTuple(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
create_named_tuple_typing = timeit('TransactionNamedTupleTyping(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
create_dataclass = timeit('TransactionDataClass(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
create_dataclass_slot = timeit('TransactionDataClassWithSlot(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())

highsmallxu

from timeit import timeit
get_obj = timeit('tans_obj.receiver', setup='tans_obj = TransactionObject(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
get_named_tuple = timeit('trans_named_tuple.receiver', setup='trans_named_tuple = TransactionNamedTuple(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
get_named_tuple_typing = timeit('trans_named_tuple_typing.receiver', setup='trans_named_tuple_typing = TransactionNamedTuple(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
get_dataclass = timeit('trans_dataclass.receiver', setup='trans_dataclass = TransactionNamedTuple(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())
get_dataclass_slot = timeit('trans_dataclass_slot.receiver', setup='trans_dataclass_slot = TransactionNamedTuple(amount=1.0, receiver="xiaoxu", date="2019-01-01", sender="jojo")', globals=globals())

highsmallxu

import io
import json
import faust
import fastavro


class AvroSchemaDecoder(faust.Schema):
  
    """An extension of Faust Schema class. The class is used by Faust when
    creating streams from Kafka topics. The decoder deserializes each message 

highsmallxu

import fastavro
import json

json_stream = faust_app.topic('my-json-topic', value_type=bytes)

# manual deseralization of JSON
@app.agent(json_stream)
async def processor(records):
    async for record in records:
        des_data = json.loads(record)

highsmallxu

class Transaction:
    def __init__(self, amount, sender, receiver, date):
      self.amount = amount
      self.sender = sender
      self.receiver = receiver
      self.date = date

highsmallxu

import collections

Transaction = collections.namedtuple('Transaction',['sender','amount','receiver','date'])

highsmallxu

record = Transaction(sender="jojo",receiver="xiaoxu",date="2020-06-08",amount=1.0)
print(record)
# Transaction(sender='jojo', amount=1.0, receiver='xiaoxu', date='2020-06-08')
print(record.receiver)
# xiaoxu