Configuration for the services the library talks to, read from the environment.
Configuration is the one place that knows which environment variable holds which setting. Nothing is read when this module is imported: the first property access loads the .env file, if there is one, and every value is read from the process environment at that moment. A caller that keeps its settings somewhere else can point the object at a different file, or export the variables itself and never let a file be loaded at all.
Typical usage example:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration() base_url = project_configuration.ubi_base_url
The settings for one process, read from the environment on first use.
Attributes:
| Name | Type | Description |
|---|---|---|
environment_file |
The str path of the |
Prepares the configuration without reading anything yet.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
environment_file
|
str | None
|
The str path of the |
None
|
load_environment_file
|
bool
|
Whether to load a |
True
|
Source code in src/tradingmachine/utilities/configuration.py
Forgets that the environment file was loaded, so the next read loads it again.
Returns:
| Type | Description |
|---|---|
None
|
None. |
Examples:
Make the configuration read the .env file again after a variable was removed from the process environment:
import os
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
print(project_configuration.ubi_base_url)
del os.environ["TRADINGMACHINE_UBI_BASE_URL"]
print(project_configuration.ubi_base_url)
project_configuration.reload()
print(project_configuration.ubi_base_url)
Pick up a variable that was added to an environment file after the first read:
import os
import tempfile
from tradingmachine.utilities import configuration
with tempfile.TemporaryDirectory() as directory:
path = os.path.join(directory, "example.env")
with open(path, "w") as environment_file:
print(
"TRADINGMACHINE_UBI_BASE_URL=http://127.0.0.1:8080",
file=environment_file,
)
project_configuration = configuration.Configuration(
environment_file=path
)
print(f"Host before: {project_configuration.mongodb_host}")
with open(path, "a") as environment_file:
print(
"TRADINGMACHINE_MONGODB_HOST=127.0.0.1",
file=environment_file,
)
project_configuration.reload()
print(f"Host after reload: {project_configuration.mongodb_host}")
Source code in src/tradingmachine/utilities/configuration.py
property
¶
The str address of the Unified Broker Interface, or None if it is not set.
Examples:
Print the address of UBI that the project's .env file names:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
print(project_configuration.ubi_base_url)
Read the address from a variable exported by the caller, without loading any file:
import os
from tradingmachine.utilities import configuration
os.environ["TRADINGMACHINE_UBI_BASE_URL"] = "http://127.0.0.1:8080"
project_configuration = configuration.Configuration(
load_environment_file=False
)
print(project_configuration.ubi_base_url)
Stop early with a clear message when the address is not configured:
property
¶
The str host MongoDB is reachable on, or None if it is not set.
Examples:
Print the host MongoDB is reached on:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
print(project_configuration.mongodb_host)
Read the host from a separate environment file, such as one for another machine:
import os
import tempfile
from tradingmachine.utilities import configuration
with tempfile.TemporaryDirectory() as directory:
path = os.path.join(directory, "other_machine.env")
with open(path, "w") as environment_file:
print(
"TRADINGMACHINE_MONGODB_HOST=192.168.1.20",
file=environment_file,
)
project_configuration = configuration.Configuration(
environment_file=path
)
print(project_configuration.mongodb_host)
property
¶
The str port MongoDB listens on, or None if it is not set.
Examples:
Print the port MongoDB listens on:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
print(project_configuration.mongodb_port)
Turn the port into a number, because every setting is read as text:
property
¶
The str name of the project's MongoDB database, or None if it is not set.
Examples:
Print the name of the project's database:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
print(project_configuration.mongodb_database_name)
Count the stored baskets in the project's database:
import pymongo
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
connection_string = project_configuration.mongodb_connection_string
database_name = project_configuration.mongodb_database_name
with pymongo.MongoClient(connection_string) as mongo_client:
collection = mongo_client[database_name]["asset_baskets"]
print(f"Stored baskets: {collection.count_documents({})}")
property
¶
The str MongoDB user to authenticate as, or None if it is not set.
Examples:
Print the user the library authenticates to MongoDB as:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
print(project_configuration.mongodb_username)
Check that a user is configured before connecting:
property
¶
The str password for the MongoDB user, or None if it is not set.
Examples:
Check that a password is configured without printing it:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
password = project_configuration.mongodb_password
print(f"A MongoDB password is set: {password is not None}")
Report the password's length only, so that it never appears in a log:
property
¶
The str MongoDB URI built from the host, port, username and password.
Examples:
Print the part of the connection string after the credentials, which is safe to show:
from tradingmachine.utilities import configuration
project_configuration = configuration.Configuration()
connection_string = project_configuration.mongodb_connection_string
print(connection_string.split("@")[1])
Connect to MongoDB with it and ask the server whether it is alive:
Example programs¶
Read settings from an environment file other than the project's .env¶
Run it with .venv/bin/python examples/utilities/configuration/configuration/alternate_environment_file.py.
"""Read settings from an environment file other than the project's `.env`.
The program writes a temporary environment file for a second machine, builds a `Configuration` that reads it, prints the settings it finds, then adds the MongoDB port to the file and calls `reload` so that the new variable is picked up. The temporary file is removed when the program ends.
A variable already set in the process environment is not replaced when a file is loaded or reloaded, which is why the program adds a variable rather than changing one.
Typical usage example:
.venv/bin/python examples/utilities/configuration/configuration/alternate_environment_file.py
"""
import os
import tempfile
from tradingmachine.utilities import configuration
class AlternateEnvironmentFile:
"""Settings read from a temporary environment file.
Attributes:
path: The str path of the environment file, or None until the program runs.
"""
def __init__(self):
"""Starts without a file, which run writes in a temporary directory.
Raises:
Nothing.
"""
self.path = None
def write_line(self, line: str) -> None:
"""Appends one `NAME=value` line to the environment file.
Args:
line: The str line to append, without a line ending.
Returns:
None.
Raises:
OSError: The file could not be written.
"""
with open(self.path, "a") as environment_file:
print(line, file=environment_file)
def print_settings(
self, project_configuration: configuration.Configuration
) -> None:
"""Prints the UBI address and the MongoDB host and port the configuration sees.
Args:
project_configuration: The tradingmachine.utilities.configuration.Configuration to read.
Returns:
None.
Raises:
Nothing.
"""
print(f" UBI base url: {project_configuration.ubi_base_url}")
print(f" MongoDB host: {project_configuration.mongodb_host}")
print(f" MongoDB port: {project_configuration.mongodb_port}")
def run(self) -> None:
"""Writes the file in a temporary directory, reads it, extends it and reloads it.
Returns:
None.
Raises:
OSError: The file could not be written.
"""
with tempfile.TemporaryDirectory() as directory:
self.path = os.path.join(directory, "second_machine.env")
self.write_line("TRADINGMACHINE_UBI_BASE_URL=http://192.168.1.20:8080")
self.write_line("TRADINGMACHINE_MONGODB_HOST=192.168.1.20")
project_configuration = configuration.Configuration(
environment_file=self.path
)
print("First read:")
self.print_settings(project_configuration)
self.write_line("TRADINGMACHINE_MONGODB_PORT=2003")
project_configuration.reload()
print("After adding the port and reloading:")
self.print_settings(project_configuration)
if __name__ == "__main__":
AlternateEnvironmentFile().run()
Report the project's settings and check that MongoDB answers with them¶
Run it with .venv/bin/python examples/utilities/configuration/configuration/configuration_report.py.
"""Report the project's settings and check that MongoDB answers with them.
The program reads every setting `Configuration` knows from the environment and the `.env` file, prints each one with the password hidden, and then connects to MongoDB with the connection string to confirm that the settings work.
Typical usage example:
.venv/bin/python examples/utilities/configuration/configuration/configuration_report.py
"""
import pymongo
from tradingmachine.utilities import configuration
class ConfigurationReport:
"""A report of the settings one process would use.
Attributes:
project_configuration: The tradingmachine.utilities.configuration.Configuration being reported.
"""
def __init__(self):
"""Creates the configuration, which reads nothing until the first setting is asked for.
Raises:
Nothing.
"""
self.project_configuration = configuration.Configuration()
def settings(self) -> dict:
"""Collects every setting, with the password replaced by a note of whether it is set.
Returns:
A dict mapping each str setting name to its str value, or to None when it is not set.
Raises:
Nothing.
"""
password_note = "not set"
if self.project_configuration.mongodb_password:
password_note = "set, hidden"
return {
"UBI base url": self.project_configuration.ubi_base_url,
"MongoDB host": self.project_configuration.mongodb_host,
"MongoDB port": self.project_configuration.mongodb_port,
"MongoDB database": self.project_configuration.mongodb_database_name,
"MongoDB user": self.project_configuration.mongodb_username,
"MongoDB password": password_note,
}
def check_mongodb(self) -> None:
"""Connects to MongoDB and prints its answer to a ping and the project's collections.
Returns:
None.
Raises:
pymongo.errors.PyMongoError: MongoDB could not be reached or refused the credentials.
"""
connection_string = self.project_configuration.mongodb_connection_string
print(f"Connecting to {connection_string.split('@')[1]}")
database_name = self.project_configuration.mongodb_database_name
with pymongo.MongoClient(connection_string) as mongo_client:
print(f"Ping: {mongo_client.admin.command('ping')}")
names = sorted(mongo_client[database_name].list_collection_names())
print(f"Collections in {database_name}: {names}")
def run(self) -> None:
"""Prints the settings, then checks MongoDB.
Returns:
None.
Raises:
pymongo.errors.PyMongoError: MongoDB could not be reached or refused the credentials.
"""
for name, value in self.settings().items():
print(f"{name:<18} {value}")
self.check_mongodb()
if __name__ == "__main__":
ConfigurationReport().run()