Skip to content

API Reference

optuna_db.db_instance

get_optuna_db()

Returns a global OptunaDatabase instance, initializing it if necessary.

If the database instance does not exist, it will be created using credentials stored in Docker secrets. If initialization fails, an exception is raised.

Returns:

Type Description
OptunaDatabase

The global OptunaDatabase instance.

Raises:

Type Description
RuntimeError

If the database initialization fails.

Source code in src/docktuna/optuna_db/db_instance.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def get_optuna_db() -> OptunaDatabase:
    """
    Returns a global OptunaDatabase instance, initializing it if necessary.

    If the database instance does not exist, it will be created using credentials
    stored in Docker secrets. If initialization fails, an exception is raised.

    Returns:
        The global OptunaDatabase instance.

    Raises:
        RuntimeError: If the database initialization fails.
    """
    global OPTUNA_DB
    if OPTUNA_DB is None:
        dotenv_path = (
            Path.home() / "project" / "docker" / "optuna_db" / "optuna_db.env"
        )
        load_dotenv(dotenv_path=dotenv_path)

        OPTUNA_DB = OptunaDatabase(
            username=getenv("OPTUNA_DB_USER"),
            db_password_secret="optuna_db_user_password",
            db_name=getenv("OPTUNA_DB_NAME"),
            hostname=getenv("OPTUNA_DB_HOST"),
        )
        try:
            _ = OPTUNA_DB.storage  # Force initialization to catch errors early
        except Exception as e:
            OPTUNA_DB = None  # Reset on failure
            raise RuntimeError(f"Failed to initialize Optuna database: {e}")
    return OPTUNA_DB

options: show_source: true

optuna_db.optuna_db.OptunaDatabase

Handles database interactions for Optuna studies, including secure retrieval of credentials from Docker secrets and connection management.

Source code in src/docktuna/optuna_db/optuna_db.py
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 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
class OptunaDatabase:
    """
    Handles database interactions for Optuna studies, including secure
    retrieval of credentials from Docker secrets and connection management.
    """

    def __init__(
        self,
        username: str,
        db_password_secret: str,
        db_name: str,
        hostname: str,
    ):
        """
        Initializes an OptunaDatabase instance with database connection details.

        Args:
            username: Database username.
            db_password_secret: Secret name for the database password.
            db_name: Name of the database.
            hostname: Database host.
        """
        self._username = username
        self._db_password_secret = db_password_secret
        self._db_name = db_name
        self._hostname = hostname

    def _read_secret(self, secret_name: str) -> str:
        """
        Reads a secret value from the Docker secrets directory.

        Args:
            secret_name: The name of the secret file to read.

        Returns:
            The contents of the secret file.

        Raises:
            Exception: If the secret file is not found.
        """
        secret_path = Path("/run/secrets") / secret_name
        try:
            with secret_path.open(mode="r") as f:
                return f.read().strip()
        except FileNotFoundError:
            raise Exception(f"Secret {secret_name} not found!")

    @property
    def username(self) -> str:
        """Retrieves the database username from secrets."""
        return self._username

    @property
    def db_name(self) -> str:
        """Retrieves the database name from secrets."""
        return self._db_name

    @property
    def hostname(self) -> str:
        """Retrieves the database hostname from secrets."""
        return self._hostname

    @property
    def _db_url(self) -> str:
        """
        Constructs a secure database connection URL dynamically
        using credentials stored in Docker secrets.

        Returns:
            The formatted database connection URL.
        """
        user = self._username
        password = self._read_secret(self._db_password_secret)
        db_name = self._db_name
        host = self._hostname

        return f"postgresql+psycopg2://{user}:{quote(password, safe='')}@{host}/{db_name}"

    @property
    def storage(self) -> RDBStorage:
        """
        Creates an Optuna RDBStorage instance for persistent study storage.

        Returns:
            The Optuna storage backend.
        """
        return RDBStorage(url=self._db_url)

    @property
    def study_summaries(self) -> list[optuna.study.StudySummary]:
        """
        Retrieves summaries of all studies stored in the database.

        Returns:
            A list of study summaries.
        """
        return optuna.study.get_all_study_summaries(storage=self.storage)

    def is_in_db(self, study_name: str) -> bool:
        """
        Checks if a study exists in the database.

        Args:
            study_name: The name of the study.

        Returns:
            True if the study exists, False otherwise.
        """
        return study_name in {
            summary.study_name for summary in self.study_summaries
        }

    def get_study_summary(self, study_name: str) -> optuna.study.StudySummary:
        """
        Retrieves the summary of a specific study.

        Args:
            study_name: The name of the study.

        Returns:
            The summary of the study.

        Raises:
            StopIteration: If the study is not found.
        """
        return next(
            summary
            for summary in self.study_summaries
            if summary.study_name == study_name
        )

    def get_best_params(self, study_name: str) -> dict[str, any]:
        """
        Retrieves the best hyperparameters from a completed study.

        Args:
            study_name: The name of the study.

        Returns:
            The best hyperparameters, or an empty dictionary if no trials exist.
        """
        study_summary = self.get_study_summary(study_name=study_name)
        if study_summary is None or study_summary.best_trial is None:
            return {}
        return study_summary.best_trial.params

    def get_study(self, study_name: str) -> optuna.Study:
        """
        Retrieves or creates an Optuna study.

        Args:
            study_name: The name of the study.

        Returns:
            The retrieved or newly created study.
        """
        with temporary_optuna_verbosity(logging_level=optuna.logging.WARNING):
            return optuna.create_study(
                study_name=study_name,
                storage=self.storage,
                load_if_exists=True,
            )

    def get_all_studies(self) -> list[optuna.Study]:
        """
        Retrieves all studies stored in the database.

        Returns:
            A list of all Optuna studies.
        """
        with temporary_optuna_verbosity(logging_level=optuna.logging.WARNING):
            return [
                self.get_study(study_name)
                for study_name in {
                    summary.study_name for summary in self.study_summaries
                }
            ]

    @property
    def num_existing_studies(self) -> int:
        """
        Returns the number of existing studies in the database.

        Returns:
            The total number of studies.
        """
        return len(self.get_all_studies())

    @staticmethod
    def get_last_update_time(study: optuna.Study) -> datetime.datetime:
        """
        Retrieves the last update time of a given study.

        Args:
            study: The study object.

        Returns:
            The timestamp of the most recent trial completion,
            or a default old date if no trials exist.
        """
        trial_complete_times = [
            trial.datetime_complete
            for trial in study.trials
            if trial.datetime_complete
        ]
        return (
            max(trial_complete_times)
            if trial_complete_times
            else datetime.datetime(1, 1, 1)
        )

    def get_latest_study(self) -> optuna.Study:
        """
        Retrieves the most recently updated study.

        Returns:
            The study with the most recent completed trial,
            or None if no studies exist.
        """
        sorted_studies = sorted(
            self.get_all_studies(),
            key=lambda x: self.get_last_update_time(x),
            reverse=True,
        )
        return sorted_studies[0] if sorted_studies else None

db_name property

Retrieves the database name from secrets.

hostname property

Retrieves the database hostname from secrets.

num_existing_studies property

Returns the number of existing studies in the database.

Returns:

Type Description
int

The total number of studies.

storage property

Creates an Optuna RDBStorage instance for persistent study storage.

Returns:

Type Description
RDBStorage

The Optuna storage backend.

study_summaries property

Retrieves summaries of all studies stored in the database.

Returns:

Type Description
list[StudySummary]

A list of study summaries.

username property

Retrieves the database username from secrets.

__init__(username, db_password_secret, db_name, hostname)

Initializes an OptunaDatabase instance with database connection details.

Parameters:

Name Type Description Default
username str

Database username.

required
db_password_secret str

Secret name for the database password.

required
db_name str

Name of the database.

required
hostname str

Database host.

required
Source code in src/docktuna/optuna_db/optuna_db.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    username: str,
    db_password_secret: str,
    db_name: str,
    hostname: str,
):
    """
    Initializes an OptunaDatabase instance with database connection details.

    Args:
        username: Database username.
        db_password_secret: Secret name for the database password.
        db_name: Name of the database.
        hostname: Database host.
    """
    self._username = username
    self._db_password_secret = db_password_secret
    self._db_name = db_name
    self._hostname = hostname

get_all_studies()

Retrieves all studies stored in the database.

Returns:

Type Description
list[Study]

A list of all Optuna studies.

Source code in src/docktuna/optuna_db/optuna_db.py
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def get_all_studies(self) -> list[optuna.Study]:
    """
    Retrieves all studies stored in the database.

    Returns:
        A list of all Optuna studies.
    """
    with temporary_optuna_verbosity(logging_level=optuna.logging.WARNING):
        return [
            self.get_study(study_name)
            for study_name in {
                summary.study_name for summary in self.study_summaries
            }
        ]

get_best_params(study_name)

Retrieves the best hyperparameters from a completed study.

Parameters:

Name Type Description Default
study_name str

The name of the study.

required

Returns:

Type Description
dict[str, any]

The best hyperparameters, or an empty dictionary if no trials exist.

Source code in src/docktuna/optuna_db/optuna_db.py
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def get_best_params(self, study_name: str) -> dict[str, any]:
    """
    Retrieves the best hyperparameters from a completed study.

    Args:
        study_name: The name of the study.

    Returns:
        The best hyperparameters, or an empty dictionary if no trials exist.
    """
    study_summary = self.get_study_summary(study_name=study_name)
    if study_summary is None or study_summary.best_trial is None:
        return {}
    return study_summary.best_trial.params

get_last_update_time(study) staticmethod

Retrieves the last update time of a given study.

Parameters:

Name Type Description Default
study Study

The study object.

required

Returns:

Type Description
datetime

The timestamp of the most recent trial completion,

datetime

or a default old date if no trials exist.

Source code in src/docktuna/optuna_db/optuna_db.py
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
@staticmethod
def get_last_update_time(study: optuna.Study) -> datetime.datetime:
    """
    Retrieves the last update time of a given study.

    Args:
        study: The study object.

    Returns:
        The timestamp of the most recent trial completion,
        or a default old date if no trials exist.
    """
    trial_complete_times = [
        trial.datetime_complete
        for trial in study.trials
        if trial.datetime_complete
    ]
    return (
        max(trial_complete_times)
        if trial_complete_times
        else datetime.datetime(1, 1, 1)
    )

get_latest_study()

Retrieves the most recently updated study.

Returns:

Type Description
Study

The study with the most recent completed trial,

Study

or None if no studies exist.

Source code in src/docktuna/optuna_db/optuna_db.py
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def get_latest_study(self) -> optuna.Study:
    """
    Retrieves the most recently updated study.

    Returns:
        The study with the most recent completed trial,
        or None if no studies exist.
    """
    sorted_studies = sorted(
        self.get_all_studies(),
        key=lambda x: self.get_last_update_time(x),
        reverse=True,
    )
    return sorted_studies[0] if sorted_studies else None

get_study(study_name)

Retrieves or creates an Optuna study.

Parameters:

Name Type Description Default
study_name str

The name of the study.

required

Returns:

Type Description
Study

The retrieved or newly created study.

Source code in src/docktuna/optuna_db/optuna_db.py
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def get_study(self, study_name: str) -> optuna.Study:
    """
    Retrieves or creates an Optuna study.

    Args:
        study_name: The name of the study.

    Returns:
        The retrieved or newly created study.
    """
    with temporary_optuna_verbosity(logging_level=optuna.logging.WARNING):
        return optuna.create_study(
            study_name=study_name,
            storage=self.storage,
            load_if_exists=True,
        )

get_study_summary(study_name)

Retrieves the summary of a specific study.

Parameters:

Name Type Description Default
study_name str

The name of the study.

required

Returns:

Type Description
StudySummary

The summary of the study.

Raises:

Type Description
StopIteration

If the study is not found.

Source code in src/docktuna/optuna_db/optuna_db.py
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def get_study_summary(self, study_name: str) -> optuna.study.StudySummary:
    """
    Retrieves the summary of a specific study.

    Args:
        study_name: The name of the study.

    Returns:
        The summary of the study.

    Raises:
        StopIteration: If the study is not found.
    """
    return next(
        summary
        for summary in self.study_summaries
        if summary.study_name == study_name
    )

is_in_db(study_name)

Checks if a study exists in the database.

Parameters:

Name Type Description Default
study_name str

The name of the study.

required

Returns:

Type Description
bool

True if the study exists, False otherwise.

Source code in src/docktuna/optuna_db/optuna_db.py
125
126
127
128
129
130
131
132
133
134
135
136
137
def is_in_db(self, study_name: str) -> bool:
    """
    Checks if a study exists in the database.

    Args:
        study_name: The name of the study.

    Returns:
        True if the study exists, False otherwise.
    """
    return study_name in {
        summary.study_name for summary in self.study_summaries
    }

options: show_source: true