How to use faust - 10 common examples
To help you get started, we’ve selected a few faust examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
marcosschroh / python-schema-registry-client / tests / serializer / test_faust_serializer.py View on Github 
def test_dumps_load_with_register_codec(client, country_schema):
payload = {"country": "Argenntina"}
country_serializer = serializer.FaustSerializer(client, "test-country", country_schema)
faust.serializers.codecs.register("country_serializer", country_serializer)
class CountryRecord(faust.Record, serializer="country_serializer"):
country: str
country_record = CountryRecord(**payload)
message_encoded = country_record.dumps()
assert message_encoded
assert len(message_encoded) > 5
assert isinstance(message_encoded, bytes)
message_decoded = CountryRecord.loads(message_encoded)
assert message_decoded == country_recorddef test_dumps_load_with_register_codec(client, country_schema):
payload = {"country": "Argenntina"}
country_serializer = serializer.FaustSerializer(client, "test-country", country_schema)
faust.serializers.codecs.register("country_serializer", country_serializer)
class CountryRecord(faust.Record, serializer="country_serializer"):
country: str
country_record = CountryRecord(**payload)
message_encoded = country_record.dumps()
assert message_encoded
assert len(message_encoded) > 5
assert isinstance(message_encoded, bytes)
message_decoded = CountryRecord.loads(message_encoded)
assert message_decoded == country_recorddef test_nested_schema_with_register_codec(client):
nested_schema = schema.AvroSchema(data_gen.NESTED_SCHENA)
order_schema = schema.AvroSchema(data_gen.ORDER_SCHENA)
customer_serializer = serializer.FaustSerializer(client, "test-nested-schema", nested_schema)
order_serializer = serializer.FaustSerializer(client, "test-order-schema", order_schema)
faust.serializers.codecs.register("customer_serializer", customer_serializer)
faust.serializers.codecs.register("order_serializer", order_serializer)
class Order(faust.Record, serializer="order_serializer"):
uid: int
class Customer(faust.Record, serializer="customer_serializer"):
name: str
uid: int
order: Order
payload = data_gen.create_nested_schema()
customer = Customer(**payload)
message_encoded = customer.dumps()def test_nested_schema_with_register_codec(client):
nested_schema = schema.AvroSchema(data_gen.NESTED_SCHENA)
order_schema = schema.AvroSchema(data_gen.ORDER_SCHENA)
customer_serializer = serializer.FaustSerializer(client, "test-nested-schema", nested_schema)
order_serializer = serializer.FaustSerializer(client, "test-order-schema", order_schema)
faust.serializers.codecs.register("customer_serializer", customer_serializer)
faust.serializers.codecs.register("order_serializer", order_serializer)
class Order(faust.Record, serializer="order_serializer"):
uid: int
class Customer(faust.Record, serializer="customer_serializer"):
name: str
uid: int
order: Order
payload = data_gen.create_nested_schema()
customer = Customer(**payload)
message_encoded = customer.dumps()
assert message_encodedfrom streaming.app import app
from streaming.config import config
from streaming.transparency.api import Records, Sources, MerkleTree
import faust
import aiohttp
class Tree(faust.Record):
size: int
source: str
# Topics
sources_topic = app.topic('ct-sources')
changed_topic = app.topic('ct-sources-changed', value_type=Tree)
cert_decoded_topic = app.topic('ct-certs-decoded')
states_table = app.Table('ct-source-states', default=int)
@app.agent(sources_topic, concurrency=50)
async def get_tree_size(sources):
base_timeout = aiohttp.ClientTimeout(total=10)
session = aiohttp.ClientSession(timeout=base_timeout)
async for source in sources:
stats = await Records(source, session).latest()
if (not source in states_table) or (stats['tree_size'] > states_table[source]):class Enrich(faust.Record):
subdomains: list = []
sinkhole: bool = False
tld: str = ''
primaryDomain: str = ''
queryValue: str = ''
queryType: str = ''
everCompromised: bool = False
tag_meta: dict = {}
classification: str = ''
tags: list = []
dynamicDns: str = ''
class Domain(faust.Record):
domain: str = ''
class Record(faust.Record):
firstSeen: str = ''
resolveType: str = ''
value: str = ''
recordHash: str = ''
lastSeen: str = ''
resolve: str = ''
source: list = []
recordType: str = ''
collected: str = ''recordHash: str = ''
lastSeen: str = ''
resolve: str = ''
source: list = []
recordType: str = ''
collected: str = ''
class PassiveDns(faust.Record):
totalRecords: int = 0
firstSeen: str = ''
lastSeen: str = ''
results: List[Record] = []
class Contact(faust.Record):
organization: str = ''
email: str = ''
name: str = ''
telephone: str = ''
class Whois(faust.Record):
tech: Contact = {}
whoisServer: str = ''
registered: str = ''
registrar: Contact = {}
domain: str = ''
registrant: str = ''
billing: dict = {}
telephone: str = ''
lastLoadedAt: str = ''def _clean_item(item: typing.Any) -> typing.Any:
if isinstance(item, Record):
return Serializer._clean_item(item.to_representation())
elif isinstance(item, str):
# str is also a sequence, need to make sure we don't iterate over it.
return item
elif isinstance(item, Mapping):
return type(item)({key: Serializer._clean_item(value) for key, value in item.items()}) # type: ignore
elif isinstance(item, Sequence):
return type(item)(Serializer._clean_item(value) for value in item) # type: ignore
return itemfrom typing import List
import faust
class Cert(faust.Record):
issuerCountry: str = ''
subjectCommonName: str = ''
subjectOrganizationName: str = ''
subjectOrganizationUnitName: str = ''
subjectGivenName: str = ''
subjectSurname: str = ''
fingerprint: str = ''
issuerStateOrProvinceName: str = ''
issuerCommonName: str = ''
subjectLocalityName: str = ''
issueDate: str = ''
subjectEmailAddress: str = ''
subjectProvince: str = ''
subjectStateOrProvinceName: str = ''
issuerEmailAddress: str = ''
subjectSerialNumber: str = ''import faust
class PageView(faust.Record):
id: str
user: str
