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

Resetting utils subtree.

Browse Source
This commit is contained in:
khushal
2024-12-12 11:47:19 +05:30
parent 9a3ef5fea4
commit 0602870bf0
132 changed files with 157 additions and 101313 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
View File
tests/kafka_test.py
+143
View File
@@ -0,0 +1,143 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 19th Sept., 2024
OBJECTIVE:
To provide a quick way to test out Kafka.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level:
import os
# Utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.queue.async_kafka import ProducerKafka, ConsumerKafka, get_ssl_context
# For async activities:
import asyncio
# Common:
from shared import constants
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
# Define the test params:
TOPIC = "kft_file_upload"
BOOTSTRAP_SERVERS = "del.ditscentre.in:9092"
SSL_CONTEXT = get_ssl_context(
ca_file = os.path.join(constants.CREDENTIALS_DIRECTORY, "kafka", "cert_authority.pem"),
cert_file = os.path.join(constants.CREDENTIALS_DIRECTORY, "kafka", "fullchain.pem"),
key_file = os.path.join(constants.CREDENTIALS_DIRECTORY, "kafka", "privkey.pem")
)
async def keep_producing():
# Create the producer:
my_producer = ProducerKafka(
topic = TOPIC,
bootstrap_servers = BOOTSTRAP_SERVERS,
security_protocol = "SSL",
ssl_context = SSL_CONTEXT
)
# Create a message for the producer to produce:
producer_message = {
"message": "Testing! Testing! 1, 2, 3...",
"ts": date_time.get_current_utc_date_time(as_string = True)
}
# Keep sending the message in intervals:
while True:
success = await my_producer.produce(producer_message)
print("PRODUCED:", success)
await asyncio.sleep(1.0)
async def keep_consuming():
# Create the consumer:
my_consumer = ConsumerKafka(
topic = TOPIC,
bootstrap_servers = BOOTSTRAP_SERVERS,
security_protocol = "SSL",
ssl_context = SSL_CONTEXT
)
# Keep checking for messages:
while True:
await asyncio.sleep(1.0)
messages = await my_consumer.consume(count = 10, timeout = 1.0)
for message in messages: print(json.to_string(message["value"]))
async def main():
await asyncio.gather(*[keep_producing(), keep_consuming()])
asyncio.run(main())
tests/mongo_test.py
+115
View File
@@ -0,0 +1,115 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 19th Sept., 2024
OBJECTIVE:
To provide a quick way to test out MongoDB.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level:
import os
# Utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.database.async_mongo_v2 import AsyncMongo
# For async activities:
import asyncio
# Common:
from shared import constants
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
async def main():
# Create an instance of the database connector:
print("Connecting...")
my_db = AsyncMongo(
connection_string = constants.MONGO_DATA_CONNECTION_STRING,
database_name = constants.MONGO_DATA_DATABASE_NAME,
max_connections = 10,
debug = True
)
await my_db.connect()
print("Connected!")
# Get some data to know if things are working well:
matching_data = await my_db.find_many(
collection = "logs",
filter = {},
sort = {"_id": -1},
limit = 3
)
print("Data:", my_db.to_json_string(matching_data))
asyncio.run(main())
tests/redis_test.py
+132
View File
@@ -0,0 +1,132 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 19th Sept., 2024
OBJECTIVE:
To provide a quick way to test out Redis.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level:
import os
# Utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.cache.async_redis_cache import AsyncRedisCache
# For async activities:
import asyncio
# Common:
from shared import constants
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
async def main():
print("Connecting...")
my_cache = AsyncRedisCache(
connection_string = r"redis://:dc4da94197c843ab6a730113c2b801d9@192.168.2.251/0",
ping_counter = 100,
debug = False
)
await my_cache.connect()
print("Connected!", end = "\n\n")
value = await my_cache.get(key = "name")
print("GET:", value)
print("TYP:", type(value), end = "\n\n")
success = await my_cache.set(
key = "name",
value = {"first": "John", "last": "Doe"},
expiry = 10
)
print("SET:", success, end = "\n\n")
value = await my_cache.get(key = "name")
print("GET:", value)
print("TYP:", type(value), end = "\n\n")
success = await my_cache.delete(key = "name")
print("DEL:", success, end = "\n\n")
value = await my_cache.get(key = "cnt")
print("GET:", value)
print("TYP:", type(value), end = "\n\n")
await my_cache.delete(key = "cnt")
for _ in range(50):
await asyncio.sleep(1.0)
counter = await my_cache.count(key = "cnt", value = 1, expiry = 10)
print("COUNTER:", counter)
asyncio.run(main())