khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241231) Too many thing happening...

Browse Source
This commit is contained in:
khushal
2024-12-31 07:34:49 +00:00
parent f67263320e
commit a07c762586
13 changed files with 406 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/queue/kafka/__init__.py
View File
utils_v2/queue/kafka/controllers/__init__.py
View File
utils_v2/queue/async_kafka.py → utils_v2/queue/kafka/controllers/async_kafka.py
+11 -10
View File
@@ -43,6 +43,9 @@ from aiokafka import AIOKafkaConsumer
from utils_v2.string import json
from utils_v2.serialization.json_serializer import JSONSerializer
# Data models:
from utils_v2.queue.kafka.models.message import ConsumedKafkaMessage
# For debugging:
from icecream import IceCreamDebugger
@@ -333,12 +336,10 @@ class ConsumerKafka:
if results:
for topic_partition, records in results.items():
for record in records:
record_dict = record.__dict__
record_dict["value"] = self.__serializer.deserialize(
data = record_dict["value"],
encoding = encoding
)
messages.append(record_dict)
messages.append(ConsumedKafkaMessage.from_aiokafka(
message = record,
deserializer = lambda x: self.__serializer.deserialize(x, encoding = encoding)
))
# Debugging print if something went wrong:
except Exception as exception: self.__printer(exception)
@@ -484,7 +485,7 @@ if __name__ == "__main__":
async def consumer_test():
consumer = ConsumerKafka(
topic = "kft_file_upload",
topic = "tick-listners",
# group_id = "assessImg",
group_id = "updateMedia",
bootstrap_servers = "wtt.ditscentre.in:9092",
@@ -497,13 +498,13 @@ if __name__ == "__main__":
while True:
messages = await consumer.consume(count = 1)
if len(messages) > 0: print("MESSAGE:", json.to_string(messages[0], default = str))
for message in messages: print(message)
await asyncio.sleep(1.0)
async def producer_test():
producer = ProducerKafka(
topic = "tickers",
topic = "tick-listners",
bootstrap_servers = "del.ditscentre.in:9092",
security_protocol = "SSL",
ssl_context = ssl_ctx
@@ -519,4 +520,4 @@ if __name__ == "__main__":
await producer.close()
asyncio.run(producer_test())
asyncio.run(consumer_test())
utils_v2/queue/kafka.py → utils_v2/queue/kafka/controllers/kafka.py
+23 -17
View File
@@ -44,6 +44,9 @@ from confluent_kafka import Consumer
from utils_v2.string import json
from utils_v2.serialization.json_serializer import JSONSerializer
# Data models:
from utils_v2.queue.kafka.models.message import ConsumedKafkaMessage
# For debugging:
from icecream import IceCreamDebugger
@@ -406,18 +409,21 @@ class ConsumerKafka:
message = self.__consumer.poll(timeout = timeout)
if message is None or message.error(): return None
else:
message_value = message.value()
if message_value is not None: message_value = self.__serializer.deserialize(message_value, encoding = encoding)
return {
"topic": message.topic(),
"partition": message.partition(),
"offset": message.offset(),
"key": message.key().decode("utf-8") if message.key() else None,
"value": message_value,
"timestamp": message.timestamp()[1],
"headers": message.headers()
}
else: return ConsumedKafkaMessage.from_confluent_kafka(
message = message,
deserializer = lambda x: self.__serializer.deserialize(x, encoding = encoding)
)
# message_value = message.value()
# if message_value is not None: message_value = self.__serializer.deserialize(message_value, encoding = encoding)
# return {
# "topic": message.topic(),
# "partition": message.partition(),
# "offset": message.offset(),
# "key": message.key().decode("utf-8") if message.key() else None,
# "value": message_value,
# "timestamp": message.timestamp()[1],
# "headers": message.headers()
# }
def consume(self, count = 1, timeout = 0.05, encoding = "utf-8"):
@@ -514,20 +520,20 @@ if __name__ == "__main__":
# Create and connect the producer:
consumer = ConsumerKafka(
topic = "tickers",
topic = "tick-listners",
config = ConsumerKafka.create_config(
bootstrap_servers = "del.ditscentre.in:9092",
group_id = "test-group",
security_protocol = "SSL",
ca_file = r"../../creds/kafka/cert_authority.pem",
cert_file = r"../../creds/kafka/fullchain.pem",
key_file = r"../../creds/kafka/privkey.pem",
ca_file = r"../../../../creds/kafka/cert_authority.pem",
cert_file = r"../../../../creds/kafka/fullchain.pem",
key_file = r"../../../../creds/kafka/privkey.pem",
)
)
consumer.connect()
print("CONSUMER READY!")
for _ in range(5):
for _ in range(500):
messages = consumer.consume(count = 3, timeout = 2.5)
print(json.to_string(messages, default = str))
utils_v2/queue/kafka/models/__init__.py
View File
utils_v2/queue/kafka/models/message.py
+257
View File
@@ -0,0 +1,257 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 30th Dec., 2024.
OBJECTIVE:
To provide a standardized structure for Kafka messages.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# For making data behaviour_models:
from pydantic import BaseModel, Field, field_validator, model_validator, AwareDatetime
from typing import Optional, Literal, Union, Dict, List, Any
# Related to Google:
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.date_time import date_time
# To work with date and time:
import datetime
import dateparser
# To make API calls:
import httpx
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class ConsumedKafkaMessage(BaseModel):
topic: str = Field(
description = "the topic on which this message was received",
frozen = True
)
partition: int = Field(
description = "the partition in which this message was received",
frozen = True
)
offset: int = Field(
description = "the message's no. in the partition",
frozen = True
)
headers: List[Any] = Field(
description = "the headers received with the message",
frozen = True
)
key: Any = Field(
description = "the key with which this message is associated; important for partition management",
frozen = True
)
value: Any = Field(
description = "the actual payload of the message",
frozen = True
)
ts: AwareDatetime | None = Field(
description = "the time at which this message was sent to the queue",
frozen = True
)
tsType: Literal[
"createTime", # ...... The time at which the producer produced the message.
"logAppendTime", # ... The time at which the message was received by the broker.
None # ............... Unknown.
] = Field(
description = "to understand the source of the timestamp",
frozen = True
)
# ┏┓ ┏•
# ┃ ┏┓┏┓╋┓┏┓
# ┗┛┗┛┛┗┛┗┗┫
# ┛
class Config:
extra = "ignore"
populate_by_name = True
# ┓┏ ┓• ┓ •
# ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓
# ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗
@field_validator("ts", mode = "before")
def parse_dates(cls, value):
if value is None: return None
if not isinstance(value, datetime.datetime):
parsed = date_time.parse_date_time(
value,
date_formats = ["%Y-%m-%d %H:%M:%S"],
timezone = None
)
value = parsed if isinstance(parsed, datetime.datetime) else dateparser.parse(value)
if isinstance(value, datetime.datetime): value = date_time.to_timezone(value, date_time.TIMEZONE_UTC)
return value
# ┏┓ •
# ┃┃┏┓┏┓┏┓┏┓┏┓╋┓┏┓┏
# ┣┛┛ ┗┛┣┛┗ ┛ ┗┗┗ ┛
# ┛
pass
# ┏┓ ┏┓
# ┃ ┓┏┏╋┏┓┏┳┓ ┣ ┓┏┏┓┏┏
# ┗┛┗┻┛┗┗┛┛┗┗ ┻ ┗┻┛┗┗┛
@staticmethod
def from_aiokafka(
message,
deserializer = None
):
"""
To populate this model directly from the output of the 'aiokafka' library.
:param message: The raw message from the library.
:param deserializer: The function to use to deserialize the contents of the message.
:return: The standardized Kafka consumed message.
"""
# Extract the key and value:
key = message.key
value = message.value
if deserializer:
if key: key = deserializer(key)
if value: value = deserializer(value)
# Build and return the model:
return ConsumedKafkaMessage(
topic = message.topic,
partition = message.partition,
offset = message.offset,
headers = message.headers or [],
key = key,
value = value,
ts = message.timestamp / 1000.0 if message.timestamp else None,
tsType = {
0: "createTime",
1: "logAppendTime"
}.get(message.timestamp_type)
)
@staticmethod
def from_confluent_kafka(
message,
deserializer = None
):
"""
To populate this model directly from the output of the 'aiokafka' library.
:param message: The raw message from the library.
:param deserializer: The function to use to deserialize the contents of the message.
:return: The standardized Kafka consumed message.
"""
# If the message was null or an error:
if message is None or message.error(): return None
# Figure out the timestamp:
raw_ts = message.timestamp()
ts_type = raw_ts[0] if raw_ts else None
ts = raw_ts[1] / 1_000.0 if raw_ts else None
# Extract the key and value:
key = message.key()
value = message.value()
if deserializer:
if key: key = deserializer(key)
if value: value = deserializer(value)
# Build and return the model:
return ConsumedKafkaMessage(
topic = message.topic(),
partition = message.partition(),
offset = message.offset(),
headers = message.headers() or [],
key = key,
value = value,
ts = ts,
tsType = {
1: "createTime",
2: "logAppendTime"
}.get(ts_type)
)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
pass
utils_v2/trading/zerodha_kite/controllers/async_zerodha_kite.py
+1 -2
View File
@@ -349,7 +349,6 @@ class AsyncZerodhaKite:
client_json = await client_response.get_json()
self.__access_token = client_json["data"]["access_token"]
self.__
print(client_response.to_markdown())
print("CLIENT RESPONSE;", json.to_string(await client_response.get_json()))
@@ -381,6 +380,6 @@ if __name__ == "__main__":
# Login flow:
print("LOGIN URL:", my_kite.login_url)
my_kite.set_request_token(input("Request Token: "))
await my_kite.get_access_token()
await my_kite.generate_session()
asyncio.run(main())