add back logs

.gitignore

@@ -1,5 +1,4 @@
 __pycache__/
-logs/
 *.pyc
 .DS_Store
 *.env

openbb_sdk/builtin_extensions/providers/model/item/registry.py

@@ -1,5 +1,9 @@
 import pkg_resources
-from openbb_provider.provider.provider_registry import Builder, ProviderRegistry
+
+from openbb_providers.openbb_provider.provider.provider_registry import (
+    Builder,
+    ProviderRegistry,
+)
 
 
 # This could get more complicated as we add more things.

openbb_sdk/builtin_extensions/providers/router/openbb_provider_router.py

@@ -6,7 +6,8 @@ from openbb_provider.model.abstract.data import Data, QueryParams
 from openbb_provider.provider.abstract.fetcher import Fetcher
 from openbb_provider.provider.abstract.provider import ProviderName
 from openbb_provider.provider.provider_map import build_provider_mapping
-from openbb_provider_plugin.model.item.registry import build_registry
+from openbb_sdk.builtin_extensions.providers.model.item.registry import build_registry
+
 from openbb_sdk_core.app.model.command_context import CommandContext
 from openbb_sdk_core.app.model.command_output import CommandOutput
 from openbb_sdk_core.app.router import Router

openbb_sdk/openbb/__init__.py

@@ -1,2 +1,4 @@
 # flake8: noqa
-from openbb_sdk_core.app.static.command_runner_wrapper import app as openbb
+from openbb_sdk.openbb_sdk_core.openbb_sdk_core.app.static.command_runner_wrapper import app
+
+account = app.account

openbb_sdk/openbb_sdk_core/openbb_sdk_core/app/static/command_runner_wrapper.py

@@ -27,7 +27,7 @@ try:
         CLASS_4ebd0208_8328_5d69_8c44_ec50939c0967,
     )
 except ImportError:
-    app = None
+    raise
 else:
 
     class CommandRunnerWrapper(CLASS_4ebd0208_8328_5d69_8c44_ec50939c0967):

openbb_sdk/openbb_sdk_core/openbb_sdk_core/app/static/package_builder.py

@@ -5,6 +5,7 @@ from json import dumps
 from pathlib import Path
 from typing import (
     Annotated,
+    Any,
     Callable,
     Dict,
     List,
@@ -17,7 +18,6 @@ from typing import (
 from uuid import NAMESPACE_DNS, uuid5
 
 import pandas as pd
-from fastapi import Query
 from starlette.routing import BaseRoute
 
 from openbb_sdk_core.app.router import RouterLoader
@@ -200,16 +200,18 @@ class MethodDefinition:
         return code
 
     @staticmethod
-    def get_type(field: Query) -> type:
-        if isclass(field.type):
-            name = field.type.__name__
+    def get_type(field: Any) -> type:
+        field_type = getattr(field, "type", Parameter.empty)
+        if isclass(field_type):
+            name = field_type.__name__
             if name.startswith("Constrained") and name.endswith("Value"):
                 name = name[11:-5].lower()
-                return getattr(builtins, name, field.type)
-        return field.type
+                return getattr(builtins, name, field_type)
+            return field_type
+        return field_type
 
     @staticmethod
-    def get_default(field: Callable):
+    def get_default(field: Any):
         field_default = getattr(field, "default", None)
         if field_default is None or field_default is MISSING:
             return Parameter.empty
@@ -220,6 +222,12 @@ class MethodDefinition:
 
         return default_default
 
+    @staticmethod
+    def is_annotated_dc(annotation) -> bool:
+        return get_origin(annotation) == Annotated and hasattr(
+            annotation.__args__[0], "__dataclass_fields__"
+        )
+
     @staticmethod
     def build_func_params(parameter_map: Dict[str, Parameter]) -> str:
         MAP = {
@@ -236,9 +244,7 @@ class MethodDefinition:
             if name == "extra_params":
                 formatted[name] = Parameter(name="kwargs", kind=Parameter.VAR_KEYWORD)
 
-            elif get_origin(param.annotation) == Annotated and isclass(
-                param.annotation.__args__[0]
-            ):
+            elif MethodDefinition.is_annotated_dc(param.annotation):
                 fields = param.annotation.__args__[0].__dataclass_fields__
                 for field_name, field in fields.items():
                     name = field_name
@@ -330,9 +336,7 @@ class MethodDefinition:
         for name, param in parameter_map.items():
             if name == "extra_params":
                 code += f"            {name} = kwargs,\n"
-            elif get_origin(param.annotation) == Annotated and isclass(
-                param.annotation.__args__[0]
-            ):
+            elif MethodDefinition.is_annotated_dc(param.annotation):
                 fields = param.annotation.__args__[0].__dataclass_fields__
                 value = {k: k for k in fields}
                 code += f"            {name} = {{"

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/formatters/formatter_with_exceptions.py

@@ -0,0 +1,181 @@
+# IMPORTATION STANDARD
+import logging
+import re
+from copy import deepcopy
+from pathlib import Path
+
+# IMPORTATION THIRDPARTY
+# IMPORTATION INTERNAL
+from openbb_sdk_core.logs.models.logging_settings import LoggingSettings
+
+
+class FormatterWithExceptions(logging.Formatter):
+    """Logging Formatter that includes formatting of Exceptions"""
+
+    DATEFORMAT = "%Y-%m-%dT%H:%M:%S%z"
+    LOGFORMAT = "%(asctime)s|%(name)s|%(funcName)s|%(lineno)s|%(message)s"
+    LOGPREFIXFORMAT = (
+        "%(levelname)s|%(appName)s|%(commitHash)s|%(appId)s|%(sessionId)s|%(userId)s|"
+    )
+
+    @staticmethod
+    def calculate_level_name(record: logging.LogRecord) -> str:
+        if record.exc_text:
+            level_name = "X"
+        elif record.levelname:
+            level_name = record.levelname[0]
+        else:
+            level_name = "U"
+
+        return level_name
+
+    @staticmethod
+    def extract_log_extra(record: logging.LogRecord):
+        log_extra = dict()
+
+        if hasattr(record, "func_name_override"):
+            record.funcName = record.func_name_override  # type: ignore
+            record.lineno = 0
+
+        if hasattr(record, "session_id"):
+            log_extra["sessionId"] = record.session_id  # type: ignore
+
+        return log_extra
+
+    @staticmethod
+    def mock_ipv4(text: str) -> str:
+        pattern = r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}"
+        replacement = " FILTERED_IP "
+        text_mocked = re.sub(pattern, replacement, text)
+
+        return text_mocked
+
+    @staticmethod
+    def mock_email(text: str) -> str:
+        pattern = r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b"
+        replacement = " FILTERED_EMAIL "
+        text_mocked = re.sub(pattern, replacement, text)
+
+        return text_mocked
+
+    @staticmethod
+    def mock_password(text: str) -> str:
+        pattern = r'("password": ")[^"]+'
+        replacement = r"\1 FILTERED_PASSWORD "
+        text_mocked = re.sub(pattern, replacement, text)
+        return text_mocked
+
+    @staticmethod
+    def mock_flair(text: str) -> str:
+        pattern = r'("FLAIR": "\[)(.*?)\]'
+        replacement = r"\1 FILTERED_FLAIR ]"
+        text_mocked = re.sub(pattern, replacement, text)
+
+        return text_mocked
+
+    @staticmethod
+    def mock_home_directory(text: str) -> str:
+        user_home_directory = str(Path.home().as_posix())
+        text_mocked = text.replace("\\\\", "/").replace(
+            user_home_directory, "MOCKING_USER_PATH"
+        )
+
+        return text_mocked
+
+    @classmethod
+    def filter_special_tags(cls, text: str) -> str:
+        text_filtered = text.replace("\n", " MOCKING_BREAKLINE ")
+        text_filtered = text_filtered.replace("'Traceback", "Traceback")
+
+        return text_filtered
+
+    @classmethod
+    def filter_piis(cls, text: str) -> str:
+        text_filtered = cls.mock_ipv4(text=text)
+        text_filtered = cls.mock_email(text=text_filtered)
+        text_filtered = cls.mock_password(text=text_filtered)
+        text_filtered = cls.mock_home_directory(text=text_filtered)
+        text_filtered = cls.mock_flair(text=text_filtered)
+
+        return text_filtered
+
+    @classmethod
+    def filter_log_line(cls, text: str):
+        text_filtered = cls.filter_special_tags(text=text)
+        text_filtered = cls.filter_piis(text=text_filtered)
+
+        return text_filtered
+
+    # OVERRIDE
+    def __init__(
+        self,
+        settings: LoggingSettings,
+        style="%",
+        validate=True,
+    ) -> None:
+        super().__init__(
+            fmt=self.LOGFORMAT,
+            datefmt=self.DATEFORMAT,
+            style=style,
+            validate=validate,
+        )
+        self.settings = settings
+
+    @property
+    def settings(self) -> LoggingSettings:
+        return deepcopy(self.__settings)
+
+    @settings.setter
+    def settings(self, settings: LoggingSettings) -> None:
+        self.__settings = settings
+
+    # OVERRIDE
+    def formatException(self, ei) -> str:
+        """Exception formatting handler
+        Parameters
+        ----------
+        ei : logging._SysExcInfoType
+            Exception to be logged
+        Returns
+        ----------
+        str
+            Formatted exception
+        """
+
+        result = super().formatException(ei)
+        return repr(result)
+
+    # OVERRIDE
+    def format(self, record: logging.LogRecord) -> str:
+        """Log formatter
+        Parameters
+        ----------
+        record : logging.LogRecord
+            Logging record
+        Returns
+        ----------
+        str
+            Formatted_log message
+        """
+
+        level_name = self.calculate_level_name(record=record)
+        log_prefix_content = {
+            "appName": self.settings.app_name,
+            "levelname": level_name,
+            "appId": self.settings.app_id,
+            "sessionId": self.settings.session_id,
+            "commitHash": self.settings.commit_hash,
+            "userId": self.settings.user_id,
+        }
+
+        log_extra = self.extract_log_extra(record=record)
+        log_prefix_content = {**log_prefix_content, **log_extra}
+        log_prefix = self.LOGPREFIXFORMAT % log_prefix_content
+
+        record.msg = record.msg.replace("|", "-MOCK_PIPE-")
+
+        log_line = super().format(record)
+        log_line = self.filter_log_line(text=log_line)
+        log_line_full = log_prefix + log_line
+
+        return log_line_full

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/handlers/path_tracking_file_handler.py

@@ -0,0 +1,79 @@
+# IMPORTATION STANDARD
+from copy import deepcopy
+from logging.handlers import TimedRotatingFileHandler
+from pathlib import Path
+
+# IMPORTATION THIRDPARTY
+# IMPORTATION INTERNAL
+from openbb_sdk_core.logs.models.logging_settings import LoggingSettings
+from openbb_sdk_core.logs.utils.expired_files import (
+    get_expired_file_list,
+    get_timestamp_from_x_days,
+    remove_file_list,
+)
+
+ARCHIVES_FOLDER_NAME = "archives"
+TMP_FOLDER_NAME = "tmp"
+
+
+class PathTrackingFileHandler(TimedRotatingFileHandler):
+    @staticmethod
+    def build_log_file_path(settings: LoggingSettings) -> Path:
+        app_name = settings.app_name
+        directory = settings.user_logs_directory
+        session_id = settings.session_id
+
+        path = directory.absolute().joinpath(f"{app_name}_{session_id}")
+        return path
+
+    def clean_expired_files(self, before_timestamp: float):
+        """Remove expired files from logs directory"""
+
+        logs_dir = self.settings.user_logs_directory
+        archives_directory = logs_dir / ARCHIVES_FOLDER_NAME
+        tmp_directory = logs_dir / TMP_FOLDER_NAME
+
+        expired_logs_file_list = get_expired_file_list(
+            directory=logs_dir,
+            before_timestamp=before_timestamp,
+        )
+        expired_archives_file_list = get_expired_file_list(
+            directory=archives_directory,
+            before_timestamp=before_timestamp,
+        )
+        expired_tmp_file_list = get_expired_file_list(
+            directory=tmp_directory,
+            before_timestamp=before_timestamp,
+        )
+        remove_file_list(file_list=expired_logs_file_list)
+        remove_file_list(file_list=expired_archives_file_list)
+        remove_file_list(file_list=expired_tmp_file_list)
+
+    @property
+    def settings(self) -> LoggingSettings:
+        return deepcopy(self.__settings)
+
+    @settings.setter
+    def settings(self, settings: LoggingSettings) -> None:
+        self.__settings = settings
+
+    # OVERRIDE
+    def __init__(
+        self,
+        settings: LoggingSettings,
+        *args,
+        **kwargs,
+    ) -> None:
+        # SETUP PARENT CLASS
+        filename = str(self.build_log_file_path(settings=settings))
+        frequency = settings.frequency
+        kwargs["when"] = frequency
+
+        super().__init__(filename, *args, **kwargs)
+
+        self.suffix += ".log"
+
+        # SETUP CURRENT CLASS
+        self.__settings = settings
+
+        self.clean_expired_files(before_timestamp=get_timestamp_from_x_days(x=5))

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/handlers/posthog_handler.py

@@ -0,0 +1,136 @@
+# IMPORT STANDARD
+import json
+import logging
+import re
+from copy import deepcopy
+from enum import Enum
+from typing import Any, Dict
+
+# IMPORT THIRD-PARTY
+from posthog import Posthog
+
+# IMPORT INTERNAL
+from openbb_sdk_core.logs.formatters.formatter_with_exceptions import (
+    FormatterWithExceptions,
+)
+from openbb_sdk_core.logs.models.logging_settings import LoggingSettings
+
+openbb_posthog = Posthog(
+    "phc_kbMqB2PNpBEkfAwQZWzbQJReplLIu8Ya6OfXcAayao9",
+    host="https://eu.posthog.com",
+)
+
+
+class PosthogHandler(logging.Handler):
+    """Posthog Handler"""
+
+    class AllowedEvents(Enum):
+        """Allowed Posthog Events"""
+
+        log_startup = "log_startup"
+        log_cmd = "log_cmd"
+        log_sdk = "log_sdk"
+        log_error = "log_error"
+        log_warning = "log_warning"
+
+    def __init__(self, settings: LoggingSettings):
+        super().__init__()
+        self._settings = settings
+        self.logged_in = False
+
+    @property
+    def settings(self) -> LoggingSettings:
+        return deepcopy(self._settings)
+
+    @settings.setter
+    def settings(self, settings: LoggingSettings) -> None:
+        self._settings = settings
+
+    def emit(self, record: logging.LogRecord):
+        try:
+            self.send(record=record)
+        except Exception:
+            self.handleError(record)
+
+    def log_to_dict(self, log_info: str) -> dict:
+        """Log to dict"""
+        log_regex = r"(STARTUP|CMD): (.*)"
+        log_dict: Dict[str, Any] = {}
+
+        for log in re.findall(log_regex, log_info):
+            log_dict[log[0]] = json.loads(log[1])
+
+        sdk_regex = r"({\"INPUT\":.*})"
+        if sdk_dict := re.findall(sdk_regex, log_info):
+            log_dict["SDK"] = json.loads(sdk_dict[0])
+
+        return log_dict
+
+    def send(self, record: logging.LogRecord):
+        """Send log record to Posthog"""
+
+        level_name = logging.getLevelName(record.levelno)
+        log_line = FormatterWithExceptions.filter_log_line(text=record.getMessage())
+
+        log_extra = self.extract_log_extra(record=record)
+        log_extra.update(dict(level=level_name, message=log_line))
+        event_name = f"log_{level_name.lower()}"
+
+        if log_dict := self.log_to_dict(log_info=log_line):
+            event_name = f"log_{list(log_dict.keys())[0].lower()}"
+            log_dict = log_dict.get("STARTUP", log_dict)
+
+            log_extra = {**log_extra, **log_dict}
+            log_extra.pop("message", None)
+
+        if re.match(r"^(QUEUE|START|END|INPUT:)", log_line) and not log_dict:
+            return
+
+        if event_name not in [e.value for e in self.AllowedEvents]:
+            return
+
+        if not self.logged_in and self._settings.user_id:
+            self.logged_in = True
+            openbb_posthog.identify(
+                self._settings.user_id,
+                {
+                    "email": self._settings.user_email,
+                    "primaryUsage": self._settings.user_primary_usage,
+                },
+            )
+            openbb_posthog.alias(self._settings.user_id, self._settings.app_id)
+
+        openbb_posthog.capture(
+            self._settings.app_id,
+            event_name,
+            properties=log_extra,
+        )
+
+    def extract_log_extra(self, record: logging.LogRecord) -> Dict[str, Any]:
+        """Extract log extra from record"""
+
+        log_extra: Dict[str, Any] = {
+            "appName": self._settings.app_name,
+            "appId": self._settings.app_id,
+            "sessionId": self._settings.session_id,
+            "commitHash": self._settings.commit_hash,
+            "platform": self._settings.platform,
+            "pythonVersion": self._settings.python_version,
+            "terminalVersion": self._settings.terminal_version,
+            "branch": self._settings.branch,
+        }
+
+        if self._settings.user_id:
+            log_extra["userId"] = self._settings.user_id
+
+        if hasattr(record, "extra"):
+            log_extra = {**log_extra, **record.extra}
+
+        if record.exc_info:
+            log_extra["exception"] = {
+                "type": str(record.exc_info[0]),
+                "value": str(record.exc_info[1]),
+                "traceback": self.format(record),
+            }
+
+        return log_extra

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/handlers_manager.py

@@ -0,0 +1,73 @@
+# IMPORT STANDARD
+import logging
+import sys
+
+# IMPORT THIRD-PARTY
+# IMPORT INTERNAL
+from openbb_sdk_core.logs.formatters.formatter_with_exceptions import (
+    FormatterWithExceptions,
+)
+from openbb_sdk_core.logs.handlers.path_tracking_file_handler import (
+    PathTrackingFileHandler,
+)
+from openbb_sdk_core.logs.handlers.posthog_handler import PosthogHandler
+from openbb_sdk_core.logs.models.logging_settings import LoggingSettings
+
+
+class HandlersManager:
+    def __init__(self, settings: LoggingSettings):
+        self.is_initialized = True
+
+        self._handlers = settings.handler_list
+        self._settings = settings
+
+        for handler_type in self._handlers:
+            if handler_type == "stdout":
+                self._add_stdout_handler()
+            elif handler_type == "stderr":
+                self._add_stderr_handler()
+            elif handler_type == "noop":
+                self._add_noop_handler()
+            elif handler_type == "file":
+                self._add_file_handler()
+            elif handler_type == "posthog":
+                self._add_posthog_handler()
+            else:
+                logging.getLogger().debug("Unknown log handler.")
+
+    def _add_posthog_handler(self):
+        handler = PosthogHandler(settings=self._settings)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        logging.getLogger().addHandler(handler)
+
+    def _add_stdout_handler(self):
+        handler = logging.StreamHandler(sys.stdout)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        logging.getLogger().addHandler(handler)
+
+    def _add_stderr_handler(self):
+        handler = logging.StreamHandler(sys.stderr)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        logging.getLogger().addHandler(handler)
+
+    def _add_noop_handler(self):
+        handler = logging.NullHandler()
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        logging.getLogger().addHandler(handler)
+
+    def _add_file_handler(self):
+        handler = PathTrackingFileHandler(settings=self._settings)
+        formatter = FormatterWithExceptions(settings=self._settings)
+        handler.setFormatter(formatter)
+        logging.getLogger().addHandler(handler)
+
+    def update_handlers(self, settings: LoggingSettings):
+        logger = logging.getLogger()
+        for hdlr in logger.handlers:
+            if isinstance(hdlr, (PathTrackingFileHandler, PosthogHandler)):
+                hdlr.settings = settings
+                hdlr.formatter.settings = settings

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/logging_manager.py

@@ -0,0 +1,136 @@
+# IMPORT STANDARD
+import json
+import logging
+from enum import Enum
+from typing import Any, Callable, Dict, Optional
+
+# IMPORT THIRD-PARTY
+from pydantic.json import pydantic_encoder
+
+from openbb_sdk_core.app.model.command_output import CommandOutput
+from openbb_sdk_core.app.model.system_settings import SystemSettings
+from openbb_sdk_core.app.model.user_settings import UserSettings
+
+# IMPORT INTERNAL
+from openbb_sdk_core.app.service.system_service import SystemService
+from openbb_sdk_core.app.service.user_service import UserService
+from openbb_sdk_core.logs.formatters.formatter_with_exceptions import (
+    FormatterWithExceptions,
+)
+from openbb_sdk_core.logs.handlers_manager import HandlersManager
+from openbb_sdk_core.logs.models.logging_settings import LoggingSettings
+
+
+class LoggingManager:
+    def __init__(
+        self,
+        system_settings: Optional[SystemSettings] = None,
+        user_settings: Optional[UserSettings] = None,
+    ) -> None:
+        self._user_settings = (
+            user_settings or UserService().read_default_user_settings()
+        )
+        self._system_settings = (
+            system_settings or SystemService().read_default_system_settings()
+        )
+        self._logging_settings = LoggingSettings(
+            user_settings=self._user_settings,
+            system_settings=self._system_settings,
+        )
+        self._handlers_manager = self._setup_handlers()
+        self._log_startup()
+
+    def _setup_handlers(self) -> HandlersManager:
+        """Setup Logging"""
+        logger = logging.getLogger(__name__)
+        logging.basicConfig(
+            level=self._logging_settings.verbosity,
+            format=FormatterWithExceptions.LOGFORMAT,
+            datefmt=FormatterWithExceptions.DATEFORMAT,
+            handlers=[],
+        )
+        handlers_manager = HandlersManager(settings=self._logging_settings)
+
+        logger.info("Logging configuration finished")
+        logger.info("Logging set to %s", self._logging_settings.handler_list)
+        logger.info("Verbosity set to %s", self._logging_settings.verbosity)
+        logger.info(
+            "LOGFORMAT: %s%s",
+            FormatterWithExceptions.LOGPREFIXFORMAT.replace("|", "-"),
+            FormatterWithExceptions.LOGFORMAT.replace("|", "-"),
+        )
+
+        return handlers_manager
+
+    def _log_startup(self) -> None:
+        def check_credentials_defined(credentials: Dict[str, Any]):
+            class CredentialsDefinition(Enum):
+                defined = "defined"
+                undefined = "undefined"
+
+            return {
+                c: CredentialsDefinition.defined.value
+                if credentials[c]
+                else CredentialsDefinition.undefined.value
+                for c in credentials
+            }
+
+        logger = logging.getLogger(__name__)
+        logger.info(
+            "STARTUP: %s ",
+            json.dumps(
+                {
+                    "PREFERENCES": self._user_settings.preferences,
+                    "KEYS": check_credentials_defined(
+                        self._user_settings.credentials.dict()
+                        if self._user_settings.credentials
+                        else {}
+                    ),
+                    "SYSTEM": self._system_settings,
+                },
+                default=pydantic_encoder,
+            ),
+        )
+
+    def log(
+        self,
+        user_settings: UserSettings,
+        system_settings: SystemSettings,
+        command_output: CommandOutput,
+        route: str,
+        func: Callable,
+        kwargs: Dict[str, Any],
+    ):
+        self._user_settings = user_settings
+        self._system_settings = system_settings
+        self._logging_settings = LoggingSettings(
+            user_settings=self._user_settings,
+            system_settings=self._system_settings,
+        )
+        self._handlers_manager.update_handlers(self._logging_settings)
+
+        if "login" in route:
+            self._log_startup()
+        else:
+            logger = logging.getLogger(__name__)
+
+            # Remove CommandContext if any
+            if "cc" in kwargs:
+                kwargs.pop("cc")
+
+            # Truncate kwargs if too long
+            kwargs = {k: str(v)[:100] for k, v in kwargs.items()}
+
+            log_level = logger.error if command_output.error else logger.info
+            log_level(
+                "ERROR: %s" if command_output.error else "CMD: %s",
+                json.dumps(
+                    {
+                        "route": route,
+                        "input": kwargs,
+                        "error": command_output.error,
+                    },
+                    default=pydantic_encoder,
+                ),
+                extra={"func_name_override": func.__name__},
+            )

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/models/logging_settings.py

@@ -0,0 +1,43 @@
+from pathlib import Path
+from typing import List, Optional
+
+from openbb_sdk_core.app.model.system_settings import SystemSettings
+from openbb_sdk_core.app.model.user_settings import UserSettings
+from openbb_sdk_core.logs.utils.utils import get_app_id, get_log_dir, get_session_id
+
+
+class LoggingSettings:
+    def __init__(self, user_settings: UserSettings, system_settings: SystemSettings):
+        user_data_directory = (
+            str(Path.home() / "OpenBBUserData")
+            if not user_settings.preferences
+            else user_settings.preferences.user_data_directory
+        )
+        hub_session = (
+            user_settings.profile.hub_session if user_settings.profile else None
+        )
+        if hub_session:
+            user_id = hub_session.user_uuid
+            user_email = hub_session.email
+            user_primary_usage = hub_session.primary_usage
+        else:
+            user_id, user_email, user_primary_usage = None, None, None
+
+        # System
+        self.commit_hash: Optional[str] = system_settings.logging_commit_hash
+        self.branch: Optional[str] = system_settings.logging_branch
+        self.app_name: str = system_settings.logging_app_name
+        self.app_id: str = get_app_id(user_data_directory)
+        self.session_id: str = get_session_id()
+        self.frequency: str = system_settings.logging_frequency
+        self.handler_list: List[str] = system_settings.logging_handlers
+        self.rolling_clock: bool = system_settings.logging_rolling_clock
+        self.verbosity: int = system_settings.logging_verbosity
+        self.platform: str = system_settings.platform
+        self.python_version: str = system_settings.python_version
+        self.terminal_version: str = system_settings.version
+        # User
+        self.user_id: Optional[str] = user_id
+        self.user_logs_directory: Path = get_log_dir(user_data_directory)
+        self.user_email: Optional[str] = user_email
+        self.user_primary_usage: Optional[str] = user_primary_usage

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/utils/expired_files.py

@@ -0,0 +1,31 @@
+# IMPORTATION STANDARD
+import contextlib
+from datetime import datetime
+from pathlib import Path
+from typing import List
+
+# IMPORTATION THIRDPARTY
+
+# IMPORTATION INTERNAL
+
+
+def get_timestamp_from_x_days(x: int) -> float:
+    timestamp_from_x_days = datetime.now().timestamp() - x * 86400
+    return timestamp_from_x_days
+
+
+def get_expired_file_list(directory: Path, before_timestamp: float) -> List[Path]:
+    expired_files = list()
+    if directory.exists and directory.is_dir():  # type: ignore
+        for file in directory.iterdir():
+            if file.is_file() and file.lstat().st_mtime < before_timestamp:
+                expired_files.append(file)
+
+    return expired_files
+
+
+def remove_file_list(file_list: List[Path]):
+    for file in file_list:
+        with contextlib.suppress(PermissionError):
+            file.unlink(missing_ok=True)
+

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/utils/system_utils.py

@@ -0,0 +1,66 @@
+# IMPORT STANDARD
+from pathlib import Path
+
+import requests
+
+# IMPORT THIRD-PARTY
+# IMPORT INTERNAL
+
+try:
+    import git
+except ImportError:
+    WITH_GIT = False
+else:
+    WITH_GIT = True
+
+GITHUB_PROJECT_URL = (
+    "https://api.github.com/repos/openbb-finance/openbbterminal/branches"
+)
+
+UNKNOWN_COMMIT_PLACEHOLDER = "unknown-commit"
+UNKNOWN_BRANCH_PLACEHOLDER = "unknown-branch"
+REPOSITORY_DIRECTORY = Path(__file__).parent.parent.parent.parent.parent
+
+
+def get_commit_hash() -> str:
+    """Get Commit Short Hash"""
+
+    git_dir = REPOSITORY_DIRECTORY.joinpath(".git")
+
+    if WITH_GIT and git_dir.is_dir():
+        repo = git.Repo(path=git_dir)
+        sha = repo.head.object.hexsha
+        short_sha = repo.git.rev_parse(sha, short=8)
+        commit_hash = f"sha:{short_sha}"
+    else:
+        commit_hash = UNKNOWN_COMMIT_PLACEHOLDER
+
+    return commit_hash
+
+
+def get_branch(commit_hash: str, request_timeout: int = 5) -> str:
+    """Get Branch Name"""
+
+    def get_branch_commit_hash(branch: str) -> str:
+        response = requests.get(
+            f"{GITHUB_PROJECT_URL}/{branch}", timeout=request_timeout
+        )
+        return "sha:" + response.json()["commit"]["sha"][:8]
+
+    for branch in ["main", "develop"]:
+        try:
+            if get_branch_commit_hash(branch) == commit_hash:
+                return branch
+        except Exception:
+            pass
+
+    git_dir = REPOSITORY_DIRECTORY.joinpath(".git")
+    if WITH_GIT and git_dir.is_dir():
+        try:
+            repo = git.Repo(path=git_dir)
+            branch = repo.active_branch.name
+            return branch
+        except Exception:
+            pass
+
+    return UNKNOWN_BRANCH_PLACEHOLDER

openbb_sdk/openbb_sdk_core/openbb_sdk_core/logs/utils/utils.py

@@ -0,0 +1,56 @@
+# IMPORT STANDARD
+import time
+import uuid
+from pathlib import Path
+
+# IMPORT THIRD-PARTY
+# IMPORT INTERNAL
+
+
+def get_session_id() -> str:
+    """UUID of the current session."""
+    session_id = str(uuid.uuid4()) + "-" + str(int(time.time()))
+    return session_id
+
+
+def get_app_id(contextual_user_data_directory: str) -> str:
+    """UUID of the current installation."""
+
+    try:
+        app_id = get_log_dir(contextual_user_data_directory).stem
+    except OSError as e:
+        if e.errno == 30:
+            print("Please move the application into a writable location.")
+            print(
+                "Note for macOS users: copy `OpenBB Terminal` folder outside the DMG."
+            )
+        else:
+            raise e
+    except Exception as e:
+        raise e
+
+    return app_id
+
+
+def get_log_dir(contextual_user_data_directory: str) -> Path:
+    """Retrieve application's log directory."""
+
+    log_dir = Path(contextual_user_data_directory).joinpath("logs").absolute()
+
+    if not log_dir.is_dir():
+        log_dir.mkdir(parents=True, exist_ok=True)
+
+    log_id = (log_dir / ".logid").absolute()
+
+    if not log_id.is_file():
+        logging_id = f"{uuid.uuid4()}"
+        log_id.write_text(logging_id, encoding="utf-8")
+    else:
+        logging_id = log_id.read_text(encoding="utf-8").rstrip()
+
+    uuid_log_dir = (log_dir / logging_id).absolute()
+
+    if not uuid_log_dir.is_dir():
+        uuid_log_dir.mkdir(parents=True, exist_ok=True)
+
+    return uuid_log_dir