db: Create a utility to get credentials from dbconfig

Create helper function that uses Augeas Shellvars to parse dbconfig-common
files.

Signed-off-by: Frederico Gomes fredericojfgomes@gmail.com
[sunil: Fix quotes not getting removed from values]
[sunil: Add test case]
Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: Sunil Mohan Adapa <sunil@medhas.org>
This commit is contained in:
Frederico Gomes 2026-02-26 16:08:05 +00:00 committed by Sunil Mohan Adapa
parent 72005d6205
commit 9a524b331b
No known key found for this signature in database
GPG Key ID: 43EA1CFF0AA7C5F2
2 changed files with 60 additions and 0 deletions

40
plinth/db/dbconfig.py Normal file
View File

@ -0,0 +1,40 @@
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Utilities for parsing dbconfig-common files with Augeas."""
import pathlib
import augeas
def get_credentials(dbconfig_path: str) -> dict[str, str]:
"""Parse dbconfig-common file with Augeas Shellvars lens."""
if not pathlib.Path(dbconfig_path).is_file():
raise FileNotFoundError(f'DB config not found: {dbconfig_path}')
aug = _load_augeas(dbconfig_path)
required = ['dbc_dbuser', 'dbc_dbpass', 'dbc_dbname']
credentials = {}
for key in required + ['dbc_dbserver']:
credentials[key] = aug.get(key).strip('\'"')
if not all(credentials.get(key) for key in required):
raise ValueError('Missing required dbconfig-common credentials')
return {
'user': credentials['dbc_dbuser'],
'password': credentials['dbc_dbpass'],
'database': credentials['dbc_dbname'],
'host': credentials['dbc_dbserver'] or 'localhost'
}
def _load_augeas(config_path: str):
"""Initialize Augeas."""
aug = augeas.Augeas(flags=augeas.Augeas.NO_LOAD +
augeas.Augeas.NO_MODL_AUTOLOAD)
pathstr = str(config_path)
aug.transform('Shellvars', pathstr)
aug.set('/augeas/context', f'/files{pathstr}')
aug.load()
return aug

View File

@ -6,6 +6,7 @@ import threading
import time
from .. import db
from ..db import dbconfig
def test_db_lock_no_wait():
@ -66,3 +67,22 @@ def test_db_lock_release():
end_time = time.time()
assert return_value
assert end_time - start_time <= 0.23
def test_dbconfig_get_credentials(tmp_path):
"""Test that parsing a dbconfig-common file works."""
file_path = tmp_path / 'test.conf'
configuration = '''
dbc_dbserver='localhost'
dbc_dbname='miniflux'
dbc_dbuser='miniflux'
dbc_dbpass='gCcNyWjyPjDH'
'''
file_path.write_text(configuration)
credentials = dbconfig.get_credentials(file_path)
assert credentials == {
'host': 'localhost',
'database': 'miniflux',
'user': 'miniflux',
'password': 'gCcNyWjyPjDH',
}