15042685551

Committed 15 May 2025 10:28AM UTC coverage: 90.504% (+0.01%) from 90.494%

Build # 15042685551

Build Type

push

github

Committed by

web-flow

Commit Message

mail channel dispatch optional (#1960)

Co-authored-by: spandan_mondal <spandan.mondal@nimblework.com>

Coverage Stats

6 of 6 new or added lines in 4 files covered. (100.0%)

2 existing lines in 1 file now uncovered.

26258 of 29013 relevant lines covered (90.5%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.7

/kairon/shared/pyscript/shared_pyscript_utils.py

from datetime import date, datetime
from typing import Text

import pytz
from dateutil.parser import isoparse
from loguru import logger
from kairon import Utility
from kairon.exceptions import AppException
from kairon.shared.actions.utils import ActionUtility
from kairon.shared.cognition.data_objects import CollectionData
from kairon.api.app.routers.bot.data import CognitionDataProcessor
cognition_processor = CognitionDataProcessor()

class PyscriptSharedUtility:

    @staticmethod
    def fetch_collection_data(query: dict):

        collection_data = CollectionData.objects(__raw__= query)

        for value in collection_data:
            final_data = {}
            item = value.to_mongo().to_dict()
            collection_name = item.pop('collection_name', None)
            is_secure = item.pop('is_secure')
            data = item.pop('data')
            data = cognition_processor.prepare_decrypted_data(data, is_secure)

            final_data["_id"] = str(item["_id"])
            final_data['collection_name'] = collection_name
            final_data['is_secure'] = is_secure
            final_data['timestamp'] = item.get("timestamp")
            final_data['data'] = data

            yield final_data

    @staticmethod
    def ensure_datetime(dt):
        if isinstance(dt, str):
            dt = isoparse(dt)
        elif isinstance(dt, date) and not isinstance(dt, datetime):
            dt = datetime.combine(dt, datetime.min.time()).replace(tzinfo=pytz.UTC)
        elif isinstance(dt, datetime):
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=pytz.UTC)
        return dt
    @staticmethod
    def get_data(collection_name: str, user: str, data_filter: dict, kwargs=None, bot: Text = None):
        if not bot:
            raise Exception("Missing bot id")

        collection_name = collection_name.lower()
        query = {"bot": bot, "collection_name": collection_name}
        start_time = kwargs.pop("start_time", None) if kwargs else None
        end_time = kwargs.pop("end_time", None) if kwargs else None

        start_time = PyscriptSharedUtility.ensure_datetime(start_time) if start_time else None
        end_time = PyscriptSharedUtility.ensure_datetime(end_time) if end_time else None
        if start_time:
            query.setdefault("timestamp", {})["$gte"] = start_time
        if end_time:
            query.setdefault("timestamp", {})["$lte"] = end_time
        if data_filter.get("raw_query"):
            query.update(data_filter.get("raw_query"))
        else:
            query.update({f"data.{key}": value for key, value in data_filter.items()})
        data = list(PyscriptSharedUtility.fetch_collection_data(query))
        return {"data": data}


    @staticmethod
    def add_data(user: str, payload: dict, bot: str = None):
        if not bot:
            raise Exception("Missing bot id")

        collection_id = cognition_processor.save_collection_data(payload, user, bot)
        return {
            "message": "Record saved!",
            "data": {"_id": collection_id}
        }


    @staticmethod
    def update_data(collection_id: str, user: str, payload: dict, bot: str = None):
        if not bot:
            raise Exception("Missing bot id")

        collection_id = cognition_processor.update_collection_data(collection_id, payload, user, bot)
        return {
            "message": "Record updated!",
            "data": {"_id": collection_id}
        }


    @staticmethod
    def delete_data(collection_id: str, user: Text, bot: Text = None):
        if not bot:
            raise Exception("Missing bot id")

        cognition_processor.delete_collection_data(collection_id, bot, user)

        return {
            "message": f"Collection with ID {collection_id} has been successfully deleted.",
            "data": {"_id": collection_id}
        }

    @staticmethod
    def delete_schedule_job(event_id: Text, bot: Text):
        if not bot:
            raise AppException("Missing bot id")

        if not event_id:
            raise AppException("Missing event id")

        logger.info(f"event: {event_id}, bot: {bot}")

        event_server = Utility.environment['events']['server_url']

        http_response = ActionUtility.execute_http_request(
            f"{event_server}/api/events/{event_id}",
            "DELETE")

        if not http_response.get("success"):
            raise AppException(http_response)
        else:
            logger.info(http_response)

1	from datetime import date, datetime	1✔
2	from typing import Text	1✔
3
4	import pytz	1✔
5	from dateutil.parser import isoparse	1✔
6	from loguru import logger	1✔
7	from kairon import Utility	1✔
8	from kairon.exceptions import AppException	1✔
9	from kairon.shared.actions.utils import ActionUtility	1✔
10	from kairon.shared.cognition.data_objects import CollectionData	1✔
11	from kairon.api.app.routers.bot.data import CognitionDataProcessor	1✔
12	cognition_processor = CognitionDataProcessor()	1✔
13
14	class PyscriptSharedUtility:	1✔
15
16	@staticmethod	1✔
17	def fetch_collection_data(query: dict):	1✔
18
19	collection_data = CollectionData.objects(__raw__= query)	1✔
20
21	for value in collection_data:	1✔
22	final_data = {}	1✔
23	item = value.to_mongo().to_dict()	1✔
24	collection_name = item.pop('collection_name', None)	1✔
25	is_secure = item.pop('is_secure')	1✔
26	data = item.pop('data')	1✔
27	data = cognition_processor.prepare_decrypted_data(data, is_secure)	1✔
28
29	final_data["_id"] = str(item["_id"])	1✔
30	final_data['collection_name'] = collection_name	1✔
31	final_data['is_secure'] = is_secure	1✔
32	final_data['timestamp'] = item.get("timestamp")	1✔
33	final_data['data'] = data	1✔
34
35	yield final_data	1✔
36
37	@staticmethod	1✔
38	def ensure_datetime(dt):	1✔
39	if isinstance(dt, str):	1✔
40	dt = isoparse(dt)	1✔
41	elif isinstance(dt, date) and not isinstance(dt, datetime):	1✔
42	dt = datetime.combine(dt, datetime.min.time()).replace(tzinfo=pytz.UTC)	1✔
43	elif isinstance(dt, datetime):	1✔
44	if dt.tzinfo is None:	1✔
45	dt = dt.replace(tzinfo=pytz.UTC)	1✔
46	return dt	1✔
47	@staticmethod	1✔
48	def get_data(collection_name: str, user: str, data_filter: dict, kwargs=None, bot: Text = None):	1✔
49	if not bot:	1✔
50	raise Exception("Missing bot id")	1✔
51
52	collection_name = collection_name.lower()	1✔
53	query = {"bot": bot, "collection_name": collection_name}	1✔
54	start_time = kwargs.pop("start_time", None) if kwargs else None	1✔
55	end_time = kwargs.pop("end_time", None) if kwargs else None	1✔
56
57	start_time = PyscriptSharedUtility.ensure_datetime(start_time) if start_time else None	1✔
58	end_time = PyscriptSharedUtility.ensure_datetime(end_time) if end_time else None	1✔
59	if start_time:	1✔
60	query.setdefault("timestamp", {})["$gte"] = start_time	1✔
61	if end_time:	1✔
UNCOV 62	query.setdefault("timestamp", {})["$lte"] = end_time	×
63	if data_filter.get("raw_query"):	1✔
UNCOV 64	query.update(data_filter.get("raw_query"))	×
65	else:
66	query.update({f"data.{key}": value for key, value in data_filter.items()})	1✔
67	data = list(PyscriptSharedUtility.fetch_collection_data(query))	1✔
68	return {"data": data}	1✔
69
70
71	@staticmethod	1✔
72	def add_data(user: str, payload: dict, bot: str = None):	1✔
73	if not bot:	1✔
74	raise Exception("Missing bot id")	1✔
75
76	collection_id = cognition_processor.save_collection_data(payload, user, bot)	1✔
77	return {	1✔
78	"message": "Record saved!",
79	"data": {"_id": collection_id}
80	}
81
82
83	@staticmethod	1✔
84	def update_data(collection_id: str, user: str, payload: dict, bot: str = None):	1✔
85	if not bot:	1✔
86	raise Exception("Missing bot id")	1✔
87
88	collection_id = cognition_processor.update_collection_data(collection_id, payload, user, bot)	1✔
89	return {	1✔
90	"message": "Record updated!",
91	"data": {"_id": collection_id}
92	}
93
94
95	@staticmethod	1✔
96	def delete_data(collection_id: str, user: Text, bot: Text = None):	1✔
97	if not bot:	1✔
98	raise Exception("Missing bot id")	1✔
99
100	cognition_processor.delete_collection_data(collection_id, bot, user)	1✔
101
102	return {	1✔
103	"message": f"Collection with ID {collection_id} has been successfully deleted.",
104	"data": {"_id": collection_id}
105	}
106
107	@staticmethod	1✔
108	def delete_schedule_job(event_id: Text, bot: Text):	1✔
109	if not bot:	1✔
110	raise AppException("Missing bot id")	1✔
111
112	if not event_id:	1✔
113	raise AppException("Missing event id")	1✔
114
115	logger.info(f"event: {event_id}, bot: {bot}")	1✔
116
117	event_server = Utility.environment['events']['server_url']	1✔
118
119	http_response = ActionUtility.execute_http_request(	1✔
120	f"{event_server}/api/events/{event_id}",
121	"DELETE")
122
123	if not http_response.get("success"):	1✔
124	raise AppException(http_response)	1✔
125	else:
126	logger.info(http_response)	1✔

digiteinfotech / kairon / 15042685551

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous