From 1e78fe5ca35c991baa943fe66f18dc9f010feefb Mon Sep 17 00:00:00 2001 From: Denis Laxalde Date: Tue, 12 Nov 2024 14:49:55 +0100 Subject: [PATCH] Define conf.Entry as a dataclass --- pgtoolkit/conf.py | 82 +++++++++++++++------------------------------- tests/test_conf.py | 52 ++++++++++++++--------------- 2 files changed, 52 insertions(+), 82 deletions(-) diff --git a/pgtoolkit/conf.py b/pgtoolkit/conf.py index 3e53062..c23d451 100644 --- a/pgtoolkit/conf.py +++ b/pgtoolkit/conf.py @@ -50,6 +50,7 @@ import sys from collections import OrderedDict from collections.abc import Iterable, Iterator +from dataclasses import KW_ONLY, MISSING, dataclass, field from datetime import timedelta from typing import IO, Any, NoReturn, Union from warnings import warn @@ -303,34 +304,27 @@ def serialize_value(value: Value) -> str: return value -class Entry: - # Holds the parsed representation of a configuration entry line. - # - # This includes the comment. +_unspecified: Any = object() - def __init__( - self, - name: str, - value: Value, - commented: bool = False, - comment: str | None = None, - raw_line: str | None = None, - ) -> None: - self._name = name - # We parse value only if not already parsed from a file - if raw_line is None and isinstance(value, str): - value = parse_value(value) - self._value = value - self.commented = commented - self.comment = comment - # Store the raw_line to track the position in the list of lines. - if raw_line is None: - raw_line = str(self) + "\n" - self.raw_line = raw_line - @property - def name(self) -> str: - return self._name +@dataclass(slots=True) +class Entry: + """Configuration entry, parsed from a line in the configuration file.""" + + name: str + _value: Value + _: KW_ONLY + commented: bool = False + comment: str | None = None + raw_line: str = field(default=_unspecified, compare=False, repr=False) + + def __post_init__(self) -> None: + if self.raw_line is _unspecified: + # We parse value only if not already parsed from a file + if isinstance(self._value, str): + self._value = parse_value(self._value) + # Store the raw_line to track the position in the list of lines. + self.raw_line = str(self) + "\n" @property def value(self) -> Value: @@ -342,24 +336,6 @@ def value(self, value: str | Value) -> None: value = parse_value(value) self._value = value - def __eq__(self, other: Any) -> bool: - if not isinstance(other, Entry): - return NotImplemented # pragma: nocover - return ( - self.name == other.name - and self.value == other.value - and self.comment == other.comment - and self.commented == other.commented - ) - - def __repr__(self) -> str: - return "<{} {}={}{}>".format( - self.__class__.__name__, - self.name, - self.value, - " (commented)" if self.commented else "", - ) - def serialize(self) -> str: return serialize_value(self.value) @@ -386,13 +362,13 @@ class EntriesProxy(dict[str, Entry]): >>> p.add('listen_addresses', '*', commented=True, comment='IP address') >>> p # doctest: +NORMALIZE_WHITESPACE - {'port': , - 'shared_buffers': , - 'listen_addresses': } + {'port': Entry(name='port', _value=5433, commented=False, comment=None), + 'shared_buffers': Entry(name='shared_buffers', _value='1GB', commented=False, comment=None), + 'listen_addresses': Entry(name='listen_addresses', _value='*', commented=True, comment='IP address')} >>> del p['shared_buffers'] >>> p # doctest: +NORMALIZE_WHITESPACE - {'port': , - 'listen_addresses': } + {'port': Entry(name='port', _value=5433, commented=False, comment=None), + 'listen_addresses': Entry(name='listen_addresses', _value='*', commented=True, comment='IP address')} Adding an existing entry fails: >>> p.add('port', 5433) @@ -551,11 +527,7 @@ def parse(self, fo: Iterable[str]) -> Iterator[tuple[pathlib.Path, IncludeType]] if not existing_entry.commented: continue self.entries[name] = Entry( - name=name, - value=value, - commented=commented, - raw_line=raw_line, - **kwargs, + name, value, commented=commented, raw_line=raw_line, **kwargs ) def parse_string(self, string: str) -> None: @@ -604,7 +576,7 @@ def __setitem__(self, key: str, value: Value) -> None: e.value = value self._update_entry(e) else: - self._add_entry(Entry(name=key, value=value)) + self._add_entry(Entry(key, value)) def get(self, key: str, default: Value | None = None) -> Value | None: try: diff --git a/tests/test_conf.py b/tests/test_conf.py index 688bf13..a4c96d2 100644 --- a/tests/test_conf.py +++ b/tests/test_conf.py @@ -307,7 +307,7 @@ def test_parse_string_include(tmp_path): def test_entry_edit(): from pgtoolkit.conf import Entry - entry = Entry(name="port", value="1234") + entry = Entry("port", "1234") assert entry.value == 1234 entry.value = "9876" assert entry.value == 9876 @@ -316,56 +316,54 @@ def test_entry_edit(): def test_entry_constructor_parse_value(): from pgtoolkit.conf import Entry - entry = Entry(name="var", value="'1.2'") + entry = Entry("var", "'1.2'") assert entry.value == "1.2" - entry = Entry(name="var", value="1234") + entry = Entry("var", "1234") assert entry.value == 1234 # If value come from the parsing of a file (ie. raw_line is provided) value should # not be parsed and be kept as is - entry = Entry(name="var", value="'1.2'", raw_line="var = '1.2'") + entry = Entry("var", "'1.2'", raw_line="var = '1.2'") assert entry.value == "'1.2'" def test_serialize_entry(): from pgtoolkit.conf import Entry - e = Entry(name="grp.setting", value=True) + e = Entry("grp.setting", True) assert "grp.setting" in repr(e) assert "grp.setting = on" == str(e) - assert "'2kB'" == Entry(name="var", value="2kB").serialize() - assert "2048" == Entry(name="var", value=2048).serialize() - assert "var = 0" == str(Entry(name="var", value=0)) - assert "var = 15" == str(Entry(name="var", value=15)) - assert "var = 0.1" == str(Entry(name="var", value=0.1)) - assert "var = 'enum'" == str(Entry(name="var", value="enum")) - assert "addrs = '*'" == str(Entry(name="addrs", value="*")) - assert "var = 'sp ced'" == str(Entry(name="var", value="sp ced")) - assert "var = 'quo''ed'" == str(Entry(name="var", value="quo'ed")) - assert "var = 'quo''ed'' and space'" == str( - Entry(name="var", value="quo'ed' and space") - ) - - assert r"'quo\'ed'" == Entry(name="var", value=r"quo\'ed").serialize() - e = Entry(name="var", value="app=''foo'' host=192.168.0.8") + assert "'2kB'" == Entry("var", "2kB").serialize() + assert "2048" == Entry("var", 2048).serialize() + assert "var = 0" == str(Entry("var", 0)) + assert "var = 15" == str(Entry("var", 15)) + assert "var = 0.1" == str(Entry("var", 0.1)) + assert "var = 'enum'" == str(Entry("var", "enum")) + assert "addrs = '*'" == str(Entry("addrs", "*")) + assert "var = 'sp ced'" == str(Entry("var", "sp ced")) + assert "var = 'quo''ed'" == str(Entry("var", "quo'ed")) + assert "var = 'quo''ed'' and space'" == str(Entry("var", "quo'ed' and space")) + + assert r"'quo\'ed'" == Entry("var", r"quo\'ed").serialize() + e = Entry("var", "app=''foo'' host=192.168.0.8") assert e.serialize() == "'app=''foo'' host=192.168.0.8'" assert str(e) == "var = 'app=''foo'' host=192.168.0.8'" e = Entry( - name="primary_conninfo", - value="port=5432 password=pa'sw0'd dbname=postgres", + "primary_conninfo", + "port=5432 password=pa'sw0'd dbname=postgres", ) assert ( str(e) == "primary_conninfo = 'port=5432 password=pa''sw0''d dbname=postgres'" ) - assert "var = 'quoted'" == str(Entry(name="var", value="'quoted'")) + assert "var = 'quoted'" == str(Entry("var", "'quoted'")) - assert "'1d'" == Entry("var", value=timedelta(days=1)).serialize() - assert "'1h'" == Entry("var", value=timedelta(minutes=60)).serialize() - assert "'61 min'" == Entry("var", value=timedelta(minutes=61)).serialize() - e = Entry("var", value=timedelta(microseconds=12000)) + assert "'1d'" == Entry("var", timedelta(days=1)).serialize() + assert "'1h'" == Entry("var", timedelta(minutes=60)).serialize() + assert "'61 min'" == Entry("var", timedelta(minutes=61)).serialize() + e = Entry("var", timedelta(microseconds=12000)) assert "'12 ms'" == e.serialize() assert " # Comment" in str(Entry("var", 1, comment="Comment"))