Skip to content

hololinked.config.Configuration

Allows to auto apply common settings used throughout the package, instead of passing these settings as arguments. Import global_config variable instead of instantiating this class. Please check global_config docstring for supported values or website documentation.

This implementation needs to be improved in general. Consider opening an issue if you have suggestions at GitHub.

Source code in hololinked/hololinked/config.py 
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
class Configuration:
    """
    Allows to auto apply common settings used throughout the package, instead of passing these settings as arguments.
    Import `global_config` variable instead of instantiating this class. Please check `global_config` docstring for supported values
    or [website documentation](https://docs.hololinked.dev/api-reference/global-config/).

    This implementation needs to be improved in general. Consider opening an issue
    if you have suggestions at [GitHub](https://github.com/hololinked-dev/hololinked/issues).
    """

    __slots__ = [
        "app_name",
        # folders
        "TEMP_DIR",
        # TCP sockets
        "TCP_SOCKET_SEARCH_START_PORT",
        "TCP_SOCKET_SEARCH_END_PORT",
        # HTTP server
        "ALLOW_CORS",
        # database
        "DB_CONFIG_FILE",
        # Eventloop
        "USE_UVLOOP",
        "TRACE_MALLOC",
        # schema validation
        "VALIDATE_SCHEMAS",
        # ZMQ
        "ZMQ_CONTEXT",
        # make debugging easier
        "DEBUG",
        # logging
        "LOG_LEVEL",
        "USE_LOG_FILE",
        "LOG_FILENAME",
        "ROTATE_LOG_FILES",
        "LOGFILE_BACKUP_COUNT",
        # "USE_STRUCTLOG",
        "COLORED_LOGS",
        # serializers
        "ALLOW_PICKLE",
        "ALLOW_UNKNOWN_SERIALIZATION",
        # internal
        "_sockets_folder",
        "_secrets_folder",
        "_logs_folder",
        "_db_folder",
    ]

    def __init__(self, app_name: str | None = None):
        self.app_name = app_name
        self._sockets_folder = "sockets"
        self._secrets_folder = "secrets"
        self._logs_folder = "logs"
        self._db_folder = "db"
        self.load_variables()

    def load_variables(self):
        """Set default values. This method is called during `__init__`"""
        # note that all variables have not been implemented yet,
        # things just come and go as of now
        self.TEMP_DIR = os.path.join(os.path.expanduser("~"), ".hololinked")
        self.TCP_SOCKET_SEARCH_START_PORT = 60000
        self.TCP_SOCKET_SEARCH_END_PORT = 65535
        self.ALLOW_CORS = False
        self.DB_CONFIG_FILE = None
        self.USE_UVLOOP = False
        self.TRACE_MALLOC = False
        # self.VALIDATE_SCHEMA_ON_CLIENT = False
        self.VALIDATE_SCHEMAS = True
        self.ZMQ_CONTEXT = zmq.asyncio.Context()
        self.DEBUG = False
        self.LOG_LEVEL = logging.DEBUG if self.DEBUG else logging.INFO
        # self.USE_STRUCTLOG = True
        self.COLORED_LOGS = False
        self.USE_LOG_FILE = False
        self.LOG_FILENAME = os.path.join(self.TEMP_DIR_LOGS, generate_main_script_log_filename(self.app_name))
        self.ROTATE_LOG_FILES = True
        self.LOGFILE_BACKUP_COUNT = 14
        # Add the filename of the main script importing this module
        self.ALLOW_PICKLE = False
        self.ALLOW_UNKNOWN_SERIALIZATION = False

        self.setup()

    def setup(self):
        """
        Actions to be done to recreate global configuration state after changing config values.
        Called after `load_variables` and `set` methods.

        Please call this method after changing config values directly specific to logging or event loop policy
        """
        self.setup_temp_dirs()

        set_global_event_loop_policy(self.USE_UVLOOP)
        if self.TRACE_MALLOC:
            tracemalloc.start()

        from .logger import setup_logging

        self.LOG_LEVEL = logging.DEBUG if self.DEBUG else self.LOG_LEVEL

        # if self.USE_STRUCTLOG: # no other option for now
        setup_logging(
            log_level=self.LOG_LEVEL,
            colored_logs=self.COLORED_LOGS,
            log_file=self.LOG_FILENAME if self.USE_LOG_FILE else None,
            rotate_log_files=self.ROTATE_LOG_FILES,
            logfile_backup_count=self.LOGFILE_BACKUP_COUNT,
        )

    def copy(self):
        """returns a copy of this config as another object"""
        other = object.__new__(Configuration)
        for item in self.__slots__:
            setattr(other, item, getattr(self, item))
        return other

    def set(self, **kwargs):
        """
        sets multiple config values at once, and recreates necessary global states.
        `load_variables` sets default values first, then overwrites with environment file values.
        This method only overwrites the specified values.
        """
        for item, value in kwargs.items():
            setattr(self, item, value)
        self.setup()

    def asdict(self):
        """returns this config as a regular dictionary"""
        return {item: getattr(self, item) for item in self.__slots__}

    def zmq_context(self) -> zmq.asyncio.Context:
        """
        Returns a global ZMQ async context. Use socket_class argument to retrieve
        a synchronous socket if necessary.
        """
        return self.ZMQ_CONTEXT

    def set_default_server_execution_context(
        self,
        invokation_timeout: int | None = None,
        execution_timeout: int | None = None,
        oneway: bool = False,
    ) -> None:
        """Sets the default server execution context for the application"""
        from .core.zmq.message import default_server_execution_context

        default_server_execution_context.invokationTimeout = invokation_timeout or 5
        default_server_execution_context.executionTimeout = execution_timeout or 5
        default_server_execution_context.oneway = oneway

    def set_default_thing_execution_context(
        self,
        fetch_execution_logs: bool = False,
    ) -> None:
        """Sets the default thing execution context for the application"""
        from .core.zmq.message import default_thing_execution_context

        default_thing_execution_context.fetchExecutionLogs = fetch_execution_logs

    @property
    def TEMP_DIR_SOCKETS(self) -> str:
        """returns the temporary directory path for IPC sockets"""
        return os.path.join(self.TEMP_DIR, self._sockets_folder)

    @property
    def TEMP_DIR_LOGS(self) -> str:
        """returns the temporary directory path for log files"""
        return os.path.join(self.TEMP_DIR, self._logs_folder)

    @property
    def TEMP_DIR_DB(self) -> str:
        """returns the temporary directory path for database files"""
        return os.path.join(self.TEMP_DIR, self._db_folder)

    @property
    def TEMP_DIR_SECRETS(self) -> str:
        """returns the temporary directory path for secret files"""
        return os.path.join(self.TEMP_DIR, self._secrets_folder)

    def setup_temp_dirs(self) -> None:
        for directory in [
            self.TEMP_DIR,
            self.TEMP_DIR_SOCKETS,
            self.TEMP_DIR_LOGS,
            self.TEMP_DIR_DB,
            self.TEMP_DIR_SECRETS,
        ]:
            try:
                os.mkdir(directory)
            except FileExistsError:
                pass
            except PermissionError:
                warnings.warn(f"permission denied to create directory {directory}", UserWarning)

    def set_temp_dir(self, path: str) -> None:
        """sets the base directory path for temporary files and application data (sockets, logs, databases, secrets)"""
        self.TEMP_DIR = path
        self.setup_temp_dirs()

    def set_sockets_folders(self, path: str) -> None:
        """sets the temporary directory path for IPC sockets"""
        self._sockets_folder = path
        self.setup_temp_dirs()

    def set_logs_folder(self, path: str) -> None:
        """sets the temporary directory path for log files"""
        self._logs_folder = path
        self.setup_temp_dirs()

    def set_db_folder(self, path: str) -> None:
        """sets the temporary directory path for database files"""
        self._db_folder = path
        self.setup_temp_dirs()

    def set_secrets_folder(self, path: str) -> None:
        """sets the temporary directory path for secret files"""
        self._secrets_folder = path
        self.setup_temp_dirs()

    def cleanup_temp_dirs(self, cleanup_databases: bool = False) -> None:
        """
        Cleans up temporary directories used by hololinked, all log files and IPC sockets are removed.
        If `cleanup_databases` is `True`, database files are also removed.
        """
        directories = [self.TEMP_DIR_SOCKETS, self.TEMP_DIR_LOGS]
        if cleanup_databases:
            directories.append(self.TEMP_DIR_DB)
        for directory in directories:
            try:
                shutil.rmtree(directory)
            except FileNotFoundError:
                pass
            except PermissionError:
                warnings.warn(f"permission denied to cleanup directory {directory}", UserWarning)

    def __del__(self):
        self.ZMQ_CONTEXT.term()

Attributes

TEMP_DIR_DB property 

TEMP_DIR_DB: str

returns the temporary directory path for database files

TEMP_DIR_LOGS property 

TEMP_DIR_LOGS: str

returns the temporary directory path for log files

TEMP_DIR_SECRETS property 

TEMP_DIR_SECRETS: str

returns the temporary directory path for secret files

TEMP_DIR_SOCKETS property 

TEMP_DIR_SOCKETS: str

returns the temporary directory path for IPC sockets

Functions

asdict 

asdict()

returns this config as a regular dictionary

Source code in hololinked/hololinked/config.py 
167
168
169
def asdict(self):
    """returns this config as a regular dictionary"""
    return {item: getattr(self, item) for item in self.__slots__}

cleanup_temp_dirs 

cleanup_temp_dirs(cleanup_databases: bool = False) -> None

Cleans up temporary directories used by hololinked, all log files and IPC sockets are removed. If cleanup_databases is True, database files are also removed.

Source code in hololinked/hololinked/config.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
def cleanup_temp_dirs(self, cleanup_databases: bool = False) -> None:
    """
    Cleans up temporary directories used by hololinked, all log files and IPC sockets are removed.
    If `cleanup_databases` is `True`, database files are also removed.
    """
    directories = [self.TEMP_DIR_SOCKETS, self.TEMP_DIR_LOGS]
    if cleanup_databases:
        directories.append(self.TEMP_DIR_DB)
    for directory in directories:
        try:
            shutil.rmtree(directory)
        except FileNotFoundError:
            pass
        except PermissionError:
            warnings.warn(f"permission denied to cleanup directory {directory}", UserWarning)

copy 

copy()

returns a copy of this config as another object

Source code in hololinked/hololinked/config.py 
150
151
152
153
154
155
def copy(self):
    """returns a copy of this config as another object"""
    other = object.__new__(Configuration)
    for item in self.__slots__:
        setattr(other, item, getattr(self, item))
    return other

load_variables 

load_variables()

Set default values. This method is called during __init__

Source code in hololinked/hololinked/config.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def load_variables(self):
    """Set default values. This method is called during `__init__`"""
    # note that all variables have not been implemented yet,
    # things just come and go as of now
    self.TEMP_DIR = os.path.join(os.path.expanduser("~"), ".hololinked")
    self.TCP_SOCKET_SEARCH_START_PORT = 60000
    self.TCP_SOCKET_SEARCH_END_PORT = 65535
    self.ALLOW_CORS = False
    self.DB_CONFIG_FILE = None
    self.USE_UVLOOP = False
    self.TRACE_MALLOC = False
    # self.VALIDATE_SCHEMA_ON_CLIENT = False
    self.VALIDATE_SCHEMAS = True
    self.ZMQ_CONTEXT = zmq.asyncio.Context()
    self.DEBUG = False
    self.LOG_LEVEL = logging.DEBUG if self.DEBUG else logging.INFO
    # self.USE_STRUCTLOG = True
    self.COLORED_LOGS = False
    self.USE_LOG_FILE = False
    self.LOG_FILENAME = os.path.join(self.TEMP_DIR_LOGS, generate_main_script_log_filename(self.app_name))
    self.ROTATE_LOG_FILES = True
    self.LOGFILE_BACKUP_COUNT = 14
    # Add the filename of the main script importing this module
    self.ALLOW_PICKLE = False
    self.ALLOW_UNKNOWN_SERIALIZATION = False

    self.setup()

set 

set(**kwargs)

sets multiple config values at once, and recreates necessary global states. load_variables sets default values first, then overwrites with environment file values. This method only overwrites the specified values.

Source code in hololinked/hololinked/config.py 
157
158
159
160
161
162
163
164
165
def set(self, **kwargs):
    """
    sets multiple config values at once, and recreates necessary global states.
    `load_variables` sets default values first, then overwrites with environment file values.
    This method only overwrites the specified values.
    """
    for item, value in kwargs.items():
        setattr(self, item, value)
    self.setup()

set_db_folder 

set_db_folder(path: str) -> None

sets the temporary directory path for database files

Source code in hololinked/hololinked/config.py 
250
251
252
253
def set_db_folder(self, path: str) -> None:
    """sets the temporary directory path for database files"""
    self._db_folder = path
    self.setup_temp_dirs()

set_default_server_execution_context 

set_default_server_execution_context(invokation_timeout: int | None = None, execution_timeout: int | None = None, oneway: bool = False) -> None

Sets the default server execution context for the application

Source code in hololinked/hololinked/config.py 
178
179
180
181
182
183
184
185
186
187
188
189
def set_default_server_execution_context(
    self,
    invokation_timeout: int | None = None,
    execution_timeout: int | None = None,
    oneway: bool = False,
) -> None:
    """Sets the default server execution context for the application"""
    from .core.zmq.message import default_server_execution_context

    default_server_execution_context.invokationTimeout = invokation_timeout or 5
    default_server_execution_context.executionTimeout = execution_timeout or 5
    default_server_execution_context.oneway = oneway

set_default_thing_execution_context 

set_default_thing_execution_context(fetch_execution_logs: bool = False) -> None

Sets the default thing execution context for the application

Source code in hololinked/hololinked/config.py 
191
192
193
194
195
196
197
198
def set_default_thing_execution_context(
    self,
    fetch_execution_logs: bool = False,
) -> None:
    """Sets the default thing execution context for the application"""
    from .core.zmq.message import default_thing_execution_context

    default_thing_execution_context.fetchExecutionLogs = fetch_execution_logs

set_logs_folder 

set_logs_folder(path: str) -> None

sets the temporary directory path for log files

Source code in hololinked/hololinked/config.py 
245
246
247
248
def set_logs_folder(self, path: str) -> None:
    """sets the temporary directory path for log files"""
    self._logs_folder = path
    self.setup_temp_dirs()

set_secrets_folder 

set_secrets_folder(path: str) -> None

sets the temporary directory path for secret files

Source code in hololinked/hololinked/config.py 
255
256
257
258
def set_secrets_folder(self, path: str) -> None:
    """sets the temporary directory path for secret files"""
    self._secrets_folder = path
    self.setup_temp_dirs()

set_sockets_folders 

set_sockets_folders(path: str) -> None

sets the temporary directory path for IPC sockets

Source code in hololinked/hololinked/config.py 
240
241
242
243
def set_sockets_folders(self, path: str) -> None:
    """sets the temporary directory path for IPC sockets"""
    self._sockets_folder = path
    self.setup_temp_dirs()

set_temp_dir 

set_temp_dir(path: str) -> None

sets the base directory path for temporary files and application data (sockets, logs, databases, secrets)

Source code in hololinked/hololinked/config.py 
235
236
237
238
def set_temp_dir(self, path: str) -> None:
    """sets the base directory path for temporary files and application data (sockets, logs, databases, secrets)"""
    self.TEMP_DIR = path
    self.setup_temp_dirs()

setup 

setup()

Actions to be done to recreate global configuration state after changing config values. Called after load_variables and set methods.

Please call this method after changing config values directly specific to logging or event loop policy

Source code in hololinked/hololinked/config.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
def setup(self):
    """
    Actions to be done to recreate global configuration state after changing config values.
    Called after `load_variables` and `set` methods.

    Please call this method after changing config values directly specific to logging or event loop policy
    """
    self.setup_temp_dirs()

    set_global_event_loop_policy(self.USE_UVLOOP)
    if self.TRACE_MALLOC:
        tracemalloc.start()

    from .logger import setup_logging

    self.LOG_LEVEL = logging.DEBUG if self.DEBUG else self.LOG_LEVEL

    # if self.USE_STRUCTLOG: # no other option for now
    setup_logging(
        log_level=self.LOG_LEVEL,
        colored_logs=self.COLORED_LOGS,
        log_file=self.LOG_FILENAME if self.USE_LOG_FILE else None,
        rotate_log_files=self.ROTATE_LOG_FILES,
        logfile_backup_count=self.LOGFILE_BACKUP_COUNT,
    )

zmq_context 

zmq_context() -> zmq.asyncio.Context

Returns a global ZMQ async context. Use socket_class argument to retrieve a synchronous socket if necessary.

Source code in hololinked/hololinked/config.py 
171
172
173
174
175
176
def zmq_context(self) -> zmq.asyncio.Context:
    """
    Returns a global ZMQ async context. Use socket_class argument to retrieve
    a synchronous socket if necessary.
    """
    return self.ZMQ_CONTEXT