abidlabs
commented
11 months ago Hi @Cozokim would you be able to correctly format the code in your example so that we can run it and reproduce it?
Closed Cozokim closed 11 months ago
abidlabs
commented
11 months ago Hi @Cozokim would you be able to correctly format the code in your example so that we can run it and reproduce it?
agg437
commented
11 months ago Hi @abidlabs, I am Ana, the colleague of Julien (@Cozokim) in LIS Data Solutions, I posted the issue before in huggingface (https://discuss.huggingface.co/t/gradio-and-logging-library-conflicts/58759 ) and I had some problems with the formatting, and probably that was happened here too when Julien created the issue. My apologies for that, I am correcting it here:
Hello there!
I am trying to combine gradio with a backend based on python classes that inherit from a parent class that allow them to have customized logs and clean the context manager after finishing the execution of the class.
This is a snippet of the error I get when trying to do so:
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\requests\sessions.py”, line 589, in request
resp = self.send(prep, **send_kwargs)
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\requests\sessions.py”, line 703, in send
r = adapter.send(request, **kwargs)
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\requests\adapters.py”, line 486, in send
resp = conn.urlopen(
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\urllib3\connectionpool.py”, line 714, in urlopen
httplib_response = self._make_request(
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\urllib3\connectionpool.py”, line 473, in make_request
log.debug(
Message: ‘%s://%s:%s “%s %s %s” %s %s’
Arguments: (‘https’, ‘api.gradio.app’, 443, ‘POST’, ‘/gradio-launched-telemetry/’, ‘HTTP/1.1’, 200, None)
— Logging error —
Traceback (most recent call last):
File "C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init.py", line 440, in format
return self.format(record)
File "C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init.py", line 436, in _format
return self._fmt % values
KeyError: ‘process_name’
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging\handlers.py”, line 73, in emit
if self.shouldRollover(record):
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging\handlers.py”, line 196, in shouldRollover
msg = “%s\n” % self.format(record)
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init_.py”, line 943, in format
return fmt.format(record)
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init_.py”, line 681, in format
s = self.formatMessage(record)
File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init_.py”, line 650, in formatMessage
return self.style.format(record)
File "C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init.py", line 442, in format
raise ValueError(‘Formatting field not found in record: %s’ % e)
ValueError: Formatting field not found in record: ‘process_name’I cannot send the full message because it is an infinite loop. You can see that is saying that cannot reach the process_name argument , this is because I am adding some extra arguments in the LogDBHandler class (I customized the class adding some arguments and logging to a DB).
This happens when I try to call the following class from main:
class SimpleGradioInterface(activities):
def init(self, process, log_level=None):
super().init(process = process, log_level = log_level)
self.interface = self.create_interface()
@staticmethod
def simple_function(word=""):
return "hi"
def create_interface(self):
interface = gr.Interface(
fn=self.simple_function,
inputs=gr.Textbox(label="Input", elem_classes="textbox"),
outputs=gr.Textbox(label="Output", elem_classes="textbox")
)
return interface
def launch(self):
self.interface.launch()Using this code in main:
if name == ‘main’:
gr_interface = SimpleGradioInterface(process=process_name, log_level='DEBUG')
gr_interface.launch()
My best guest is there is some kind of conflict between python context manager definition, the definition I have in my parent class and the one defined in gradio, because if I do this test, it seems possible the combination with the backend logs:
class HitActivity(activities):
@log_decorator()
def say_hello(self):
print(“Hello from HitActivity!”)
return “Greeting done!”
def wrapped_hit_activity_function(_=None):
activity = HitActivity()
try:
activity.enter()
result = activity.say_hello()
finally:
activity.exit(None, None, None)
return result
interface = gr.Interface(
fn=wrapped_hit_activity_function,
inputs=gr.Textbox(label="Input", elem_classes="textbox"),
outputs=gr.Textbox(label="Output", elem_classes="textbox")
)
interface.launch()And then I get the expected output:
2023-10-16 17:38:36,432 - INFO - RPA Highlighter - HitActivity - Running: [Level 1] HitActivity is running - anagongue
2023-10-16 17:38:36,433 - DEBUG - RPA Highlighter - HitActivity - Running: [Level 1] Method say_hello running - anagongue
2023-10-16 17:38:36,435 - DEBUG - RPA Highlighter - HitActivity - Ended: [Level 1] Method say_hello ended successfully - anagongue
2023-10-16 17:38:36,437 - INFO - RPA Highlighter - HitActivity - Ended: [Level 1] HitActivity ended - anagongueIs there some kind of consideration I should have about how gradio uses the python context manager? In my case I am using this code for defining it in the parent class and make the child classes inherit:
# !/usr/bin/env python
# coding: utf-8
# import config variables
# Database
from bin.config import db_host, db_port, db_name, db_user, db_pwd, col_logs, log_size, process_name
from bin import setup_logger
# Mail
from bin import MailException, EmailNotifier
from bin.config import mail_host, mail_port, sender, mail_pass, recipient
import traceback # error formatting
import sys
from activities.custom_exceptions import CriticalException, WarningException, GeneralException
# Manage threads for activities, force to use one thread at a time when executing
# an activity (class), see __exit__ method in activities parent class
from threading import Lock
# LoggingMixin and EmailNotificationMixin do not need __init__ method because
# for being mixins, the only objective is to encapsulate funcionalities inside a
# class. They are not instances of their own and their use is dependant of the
# activities class.
class LoggingMixin:
'''
Mixin to Encapsulate behaviour of logging.
'''
DEFAULT_LOG_LEVEL = "DEBUG"
def configure_logger(self, log_level=None):
# Use the provided log_level or fallback to the default
effective_log_level = log_level or self.DEFAULT_LOG_LEVEL
# Initialize logger
self.logger = setup_logger(
db_host,
db_port,
db_name,
db_user,
db_pwd,
col_logs,
log_size,
effective_log_level,
process_name
)
def log_start(self):
nesting_level = len(activities._activity_stack) # get level of activity
self.logger.info(f"[Level {nesting_level}] {self.activity} started",
extra={
'activity': self.activity,
'status': "Start",
'process_name': self.process
})
def log_running(self):
nesting_level = len(activities._activity_stack)
self.logger.info(f"[Level {nesting_level}] {self.activity} is running",
extra={
'activity': self.activity,
'status': "Running",
'process_name': self.process
})
def log_end(self, exc_type=None, exc_val=None, exc_tb=None):
nesting_level = len(activities._activity_stack)
self.logger.info(f"[Level {nesting_level}] {self.activity} ended",
extra={
'activity': self.activity,
'status': "Ended",
'process_name': self.process
})
class EmailNotificationMixin:
'''
Mixin to encapsulate behaviour of mailing
'''
def send_error_notification(self, exc_type, exc_val, exc_tb):
# Format the traceback
formatted_traceback = traceback.format_exception(
exc_type, exc_val, exc_tb)
# Send email notification about the error
notifier = EmailNotifier(mail_host, mail_port,
sender, recipient, mail_pass)
notifier.send_mail(
process_name=self.process,
message='',
exception={
'activity': self.activity,
'detail': f"{''.join(formatted_traceback)}"
},
files_to_send=['output/log_file.log']
)
def get_custom_msg(exception_obj, default_message=None):
'''
Customize exception message with extra info or just add nothing
'''
return exception_obj.custom_message or default_message or ''
def log_decorator(custom_message=None, info_callback = None):
'''
Decorator used to give a default format to activities methods logs for the levels:
- DEBUG.
- WARNING.
- ERROR.
- CRITICAL.
(INFO level is reserved for pointing out the three main status of the activities:
Start, Runnning and Ended. It is already defined in LogginMixin class)
'''
def decorator(method):
def wrapper(self, *args, **kwargs):
nesting_level = len(activities._activity_stack)
self.logger.debug(f"[Level {nesting_level}] Method {method.__name__} running",
extra={
'activity': self.activity,
'status': "Running",
'process_name': self.process
})
result = None
try:
#pdb.set_trace()
result = method(self, *args, **kwargs)
# Log dynamic message based on method's result
if info_callback:
message = info_callback(result, nesting_level)
self.logger.info(message,
extra={
'activity': self.activity,
'status': "Running",
'process_name': self.process
})
self.logger.debug(f"[Level {nesting_level}] Method {method.__name__} ended successfully",
extra={
'activity': self.activity,
'status': "Ended",
'process_name': self.process
})
#pdb.set_trace()
return result
except CriticalException as ce: # Handle critical exceptions
# Handle custom message from the exception
print('criticalexception')
custom_msg = get_custom_msg(ce, custom_message)
self.logger.critical(f"[Level {nesting_level}] Method {method.__name__} encountered a critical error: {custom_msg}{str(ce)}",
extra={
'activity': self.activity,
'status': "Ended",
'process_name': self.process
})
raise ce
except WarningException as we: # warnings, it is not error but it may be different from the expected outcome
print('warningexception')
custom_msg = get_custom_msg(we, custom_message)
self.logger.warning(f"[Level {nesting_level}] Method {method.__name__} had a warning: {custom_msg}{str(we)}",
extra={
'activity': self.activity,
'status': "Ended",
'process_name': self.process
})
raise we
except GeneralException as ge: # handle general errors
print('generalexception')
custom_msg = get_custom_msg(ge, custom_message)
self.logger.error(f"[Level {nesting_level}] Method {method.__name__} encounter an error: {custom_msg} {str(ge)}",
extra={
'activity': self.activity,
'status': "Ended",
'process_name': self.process
})
raise ge
return wrapper
return decorator
class activities(LoggingMixin, EmailNotificationMixin):
'''
activities class is the parent class of
the set of activities
that forms the RPA process.
It inherits from Mixins LogginMixin and EmailNotificationMixin.
All activities inherit the same arguments:
process_name, activity (activity name) and status (Start,
Running, Ended)
'''
_activity_stack = [] # list to be used to stack activities
# and keep a record of current activities in use.
# This list allows us to organize activity log sublevels
_stack_lock = Lock() # lock object
def __init__(self, process = process_name, log_level=None):
self.process = process # RPA process name
# Activity name of the RPA process, step of the process
# automatically set to the name of the class
self.activity = self.__class__.__name__
self.configure_logger(log_level)
def __enter__(self):
with self._stack_lock: # Acquire the lock
self._activity_stack.append(self.activity) # add activity to the stack
self.log_start()
self.log_running()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if exc_type and issubclass(exc_type, CriticalException): # Check if it's a critical exception
self.send_error_notification(exc_type, exc_val, exc_tb)
with self._stack_lock: # Acquire the lock
self._activity_stack.pop() # remove activity from stack to release resources
sys.exit(f"Terminating due to critical error: {exc_val}")
elif exc_type and issubclass(exc_type, GeneralException):
self.send_error_notification(exc_type, exc_val, exc_tb)
with self._stack_lock: # Acquire the lock
self._activity_stack.pop() # remove activity from stack to release resources
elif exc_type and issubclass(exc_type, WarningException):
with self._stack_lock: # Acquire the lock
self._activity_stack.pop() # remove activity from stack to release resources
else:
self.log_end(exc_type, exc_val, exc_tb)
with self._stack_lock: # Acquire the lock
self._activity_stack.pop() # remove activity from stack to release resources
Another code you are going to need is the logger_setup() function :
#!/usr/bin/env python
# coding: utf-8
# logger_setup
# logging
import pdb
import logging
import logging.config
import traceback
# time tracking
import time
from datetime import datetime
# json formatting
import json
from jsonschema import Draft7Validator, ValidationError
# MongoDB specific imports
from pymongo.errors import ConnectionFailure, OperationFailure, PyMongoError
# LogDBHandler class
from handlers import LogDBHandler
def setup_logger(db_host, db_port, db_name, db_user, db_pwd ,col_logs, log_size,
log_level, process_name):
with open('config/LOG_CONFIG.json', 'r') as f:
CONFIG = json.load(f)
for handler_name in CONFIG['handlers']:
if CONFIG['handlers'][handler_name]['class'] == 'logging.handlers.RotatingFileHandler':
# Update log size value
CONFIG['handlers'][handler_name]['maxBytes'] = log_size
logger_handler = LogDBHandler(
process_name=process_name,
CONFIG=CONFIG,
db_host=db_host,
db_port=db_port,
db_name=db_name,
db_collection_name=col_logs,
user=db_user,
db_pwd = db_pwd,
col_logs_size=log_size,
loglevel=log_level)
# Determine which logger to use based on MongoDB connection status
if logger_handler.db_connected:
logger_name = 'loggerDB'
else:
logger_name = 'DBerror'
logger = logging.getLogger(logger_name)
if not any(isinstance(handler, LogDBHandler) for handler in logger.handlers):
logger.addHandler(logger_handler)
print("Logger Name:", logger.name)
print("Logger Handlers:", logger.handlers)
return logger
And the customized class LogDBHandler
#!/usr/bin/env python
# coding: utf-8
# logdb_handler
# logging
import pdb
import logging
import logging.config
import traceback
# time tracking
import time
from datetime import datetime
# json formatting
import json
from jsonschema import Draft7Validator, ValidationError
# MongoDB specific imports
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, OperationFailure, PyMongoError
class LogDBHandler(logging.Handler):
'''
Customized logging handler that puts logs to the MongoDB database.
pymongo required
'''
def __init__(self, process_name, CONFIG, db_host, db_port, db_name,
db_collection_name, user, db_pwd, col_logs_size, loglevel):
'''
init method for class.
Params:
----------
CONFIG: dictionary with the log configuration. See config/LOG_CONFIJ.json
db_host: ip host
db_port: port
db_name: database name to connect
db_collection_name: collection to insert documents
user: user implied
db_pwd: database password
col_logs_size: maximum log size
log_level: level in which logging must start
(it only affects for logging in DB. If logging in DB fails the starting log level
is warning by default)
Output:
----------
Customized handler
'''
logging.Handler.__init__(self) # instance of LogRecord class.
# MongoDB attributes
self.db_host = db_host
self.db_port = db_port
self.db_name = db_name
self.db_collection_name = db_collection_name
self.col_logs_size = col_logs_size
self.user = user
self.db_pwd = db_pwd
self.db_connected = False # MongoDB connection flag
# Logging attributes
self.process_name = process_name
# schema data validation
with open('config/validate_schema.json', 'r') as f:
self.schema_data = json.load(f)
# File logger for internal errors MongoDB related
# it must be initialized a specific logger for
# DB errors to avoid infinite recursions
# and dependencies between this logger and loggerDB.
self.DBerror = self.log_level(CONFIG, loglevel, f"{self.user}_DBerror")
# Log handling
# Try connecting to MongoDB
try:
self.client=self._connect_to_mongo()
self.client.admin.command('ping') # will throw an exception if no connection
self.db_connected = True
# Check collection exists
if not self.collection_exists():
self.create_collection(col_logs_size)
# Log in MongoDB
self.loggerDB = self.log_level(CONFIG,
loglevel,
self.user) # only create loggerDB if connection available
self.loggerDB.propagate = False # set to false to avoid recursion error
except Exception as e:
self.DBerror.error(f"An Exception occurred: {e}",
extra={'process_name': self.process_name,
'activity': 'Database architecture',
'status': "Ended", 'user':f"{self.user}_DBerror"})
self.DBerror.warning(
"Skipping MongoDB log as connection is not active.",
extra={'process_name': self.process_name,'activity': 'Database architecture',
'status': "Ended",'user':f"{self.user}_DBerror"})
def _connect_to_mongo(self):
'''
Connects to MongoDB and returns the client.
'''
timeout = 5000 # Timeout in milliseconds (5 seconds)
client = MongoClient(
f"{self.db_host}:{self.db_port}",
username=self.user,
password=self.db_pwd,
authsource="admin",
authMechanism="DEFAULT",
serverSelectionTimeoutMS=timeout
)
return client
def collection_exists(self):
"""
Check if the collection exists in the database.
"""
db = self.client[self.db_name]
return self.db_collection_name in db.list_collection_names()
def create_collection(self, col_logs_size):
"""
Creates collection in MongoDB
"""
db = self.client[self.db_name]
# Size limits settings specified with optional arguments
# in capped_options
capped_options = {
"capped": True,
"size": col_logs_size # Size in bytes
}
db.create_collection(
self.db_collection_name,
validator=self.schema_data,
**capped_options
)
def log_level(self, CONFIG, loglevel, user):
'''
Define log level to start with while logging
and format logs.
'''
#pdb.set_trace()
# Load configuration definition for logging (levels and files) with JSON
logging.config.dictConfig(CONFIG)
# Initialize logger
logger = logging.getLogger(user)
logger.setLevel(loglevel)
logger.addHandler(self)
logger.propagate = False
#pdb.set_trace()
return logger
def emit(self, record): # it is used to format the record,
# record is the default parameter
# of the method in logging library
'''
Create and format log message.
Log in MongoDB and local log.
'''
#pdb.set_trace()
# Set current time
tm_str = time.strftime("%Y-%m-%d %H:%M:%S",
time.localtime(record.created))
tm = datetime.strptime(tm_str, '%Y-%m-%d %H:%M:%S')
# Clear the log message so it can be put to db via MongoDB
# (escape quotes)
self.log_msg = record.msg
self.log_msg = self.log_msg.strip()
self.log_msg = self.log_msg.replace('\'', '\'\'')
# Add attributes to record class
record.process_name = self.process_name
record.user = self.user
# Create a document to insert into MongoDB
status = getattr(record, 'status', 'Ended')
if status not in ['Start', 'Running', 'Ended']:
raise ValueError(
'''Invalid status value provided.
It must be one of ["Start", "Running", "Ended"].''')
log_document = {
'process_name' : record.process_name,
'activity': getattr(record, 'activity', None),
'timestamp': tm,
'resource': getattr(record, 'resource', self.user),
'log_level': record.levelname,
'log_message': self.log_msg,
'status': status,
'user': self.user
}
self.log_MongoDB(record, log_document)
#pdb.set_trace()
def insert_log_document(self, record, log_document):
"""
Insert the log document into the collection.
"""
db = self.client[self.db_name]
collection = db[self.db_collection_name]
try:
collection.insert_one(log_document)
except Exception as e:
self.DBerror.error(f"Failed to insert log document: {e}",
extra={'process_name': self.process_name,
'activity': 'Database architecture',
'status': "Ended", 'user': f"{self.user}_DBerror"})
def log_MongoDB(self, record, log_document):
'''
Logging in MongoDB.
'''
self.insert_log_document(record, log_document)
You can omit email notification code, it is not necessary for this test, if there is something I can explain better tell me please, Best and thanks in advance, Ana
abidlabs
commented
11 months ago Hi @agg437 thanks for posting the code. But its quite overwhelming to wade through all of this code. Can you please provide a minimal code example that we can use to reproduce the issue above? See: https://stackoverflow.com/help/minimal-reproducible-example
abidlabs
commented
11 months ago Closing for lack of follow up. If we have a repro, we can reopen.
Describe the bug
Hello there!
I am trying to combine gradio with a backend based on python classes that inherit from a parent class that allow them to have customized logs and clean the context manager after finishing the execution of the class.
This is a snippet of the error I get when trying to do so:
`` File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\requests\sessions.py”, line 589, in request resp = self.send(prep, send_kwargs) File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\requests\sessions.py”, line 703, in send r = adapter.send(request, kwargs) File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\requests\adapters.py”, line 486, in send resp = conn.urlopen( File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\urllib3\connectionpool.py”, line 714, in urlopen httplib_response = self._make_request( File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\site-packages\urllib3\connectionpool.py”, line 473, in make_request log.debug( Message: ‘%s://%s:%s “%s %s %s” %s %s’ Arguments: (‘https’, ‘api.gradio.app’, 443, ‘POST’, ‘/gradio-launched-telemetry/’, ‘HTTP/1.1’, 200, None) — Logging error — Traceback (most recent call last): File "C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init.py", line 440, in format return self.format(record) File "C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init.py", line 436, in _format return self._fmt % values KeyError: ‘process_name’
During handling of the above exception, another exception occurred:
Traceback (most recent call last): File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging\handlers.py”, line 73, in emit if self.shouldRollover(record): File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging\handlers.py”, line 196, in shouldRollover msg = “%s\n” % self.format(record) File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logginginit.py”, line 943, in format return fmt.format(record) File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logginginit.py”, line 681, in format s = self.formatMessage(record) File “C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logginginit.py”, line 650, in formatMessage return self.style.format(record) File "C:\Users\AnaGonzalezGuerra\anaconda3\envs\RPA\lib\logging_init.py", line 442, in format raise ValueError(‘Formatting field not found in record: %s’ % e) ValueError: Formatting field not found in record: ‘process_name’
`` I cannot send the full message because it is an infinite loop. You can see that is saying that cannot reach the process_name argument , this is because I am adding some extra arguments in the LogDBHandler class (I customized the class adding some arguments and logging to a DB).
This happens when I try to call the following class from main:
`` class SimpleGradioInterface(activities): def init(self, process, log_level=None): super().init(process = process, log_level = log_level) # Call the parent constructor self.interface = self.create_interface() ''' @staticmethod def simple_function(word=""): return "hi"
def create_interface(self):
def launch(self):
`` Using this code in main:
`` if name == ‘main’:
gr_interface = SimpleGradioInterface(process=process_name, log_level='DEBUG') gr_interface.launch() ``
My best guest is there is some kind of conflict between python context manager definition, the definition I have in my parent class and the one defined in gradio, because if I do this test it seems possible the combination with the backend logs:
`` class HitActivity(activities): @log_decorator() def say_hello(self): print(“Hello from HitActivity!”) return “Greeting done!”
def wrapped_hit_activityfunction(=None): activity = HitActivity() try: activity.enter() result = activity.say_hello() finally: activity.exit(None, None, None) return result
interface = gr.Interface(
interface.launch() `` And then I get the expected output:
`` 2023-10-16 17:38:36,432 - INFO - RPA Highlighter - HitActivity - Running: [Level 1] HitActivity is running - anagongue
2023-10-16 17:38:36,433 - DEBUG - RPA Highlighter - HitActivity - Running: [Level 1] Method say_hello running - anagongue
2023-10-16 17:38:36,435 - DEBUG - RPA Highlighter - HitActivity - Ended: [Level 1] Method say_hello ended successfully - anagongue
2023-10-16 17:38:36,437 - INFO - RPA Highlighter - HitActivity - Ended: [Level 1] HitActivity ended - anagongue ''' Is there some kind of consideration I should have about how gradio uses the python context manager? In my case I am using this code for defining it in the parent class and make the child classes inherit:
''' class activities(LoggingMixin, EmailNotificationMixin): ''' activities class is the parent class of the set of activities that forms the RPA process. It inherits from Mixins LogginMixin and EmailNotificationMixin. All activities inherit the same arguments: process_name, activity (activity name) and status (Start, Running, Ended)
'''
_activity_stack = [] # list to be used to stack activities
and keep a record of current activities in use.
_stack_lock = Lock() # lock object
def init(self, process = process_name, log_level=None): self.process = process # RPA process name
Activity name of the RPA process, step of the process
def enter(self): with self._stack_lock: # Acquire the lock self._activity_stack.append(self.activity) # add activity to the stack self.log_start() self.log_running() return self
def exit(self, exc_type, exc_val, exc_tb):
``
Thanks for your help,
Have you searched existing issues? 🔎
Reproduction
See above
Screenshot
No response
Logs
No response
System Info
Severity
I can work around it