mirror of
https://github.com/enpaul/keyosk.git
synced 2024-11-05 06:07:06 +00:00
Add initial configuration module with storage settings
Add fields module for custom de/serializer fields
This commit is contained in:
parent
f0b1217da7
commit
2e3c10209b
23
keyosk/config/__init__.py
Normal file
23
keyosk/config/__init__.py
Normal file
@ -0,0 +1,23 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
import marshmallow as msh
|
||||
|
||||
from keyosk.config.storage import KeyoskStorageConfig
|
||||
from keyosk.config.storage import StorageConfigSerializer
|
||||
|
||||
|
||||
@dataclass
|
||||
class KeyoskConfig:
|
||||
"""Configuration storage"""
|
||||
|
||||
storage: KeyoskStorageConfig = KeyoskStorageConfig()
|
||||
|
||||
|
||||
class ConfigSerializer(msh.Schema):
|
||||
"""De/serializer for the application configuration
|
||||
|
||||
Fields on this class map 1:1 with the dataclass parameters on the
|
||||
:class:`KeyoskConfig` class.
|
||||
"""
|
||||
|
||||
storage = msh.fields.Nested(StorageConfigSerializer)
|
155
keyosk/config/storage.py
Normal file
155
keyosk/config/storage.py
Normal file
@ -0,0 +1,155 @@
|
||||
from dataclasses import asdict
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Dict
|
||||
from typing import Mapping
|
||||
from typing import Optional
|
||||
from typing import Union
|
||||
|
||||
import marshmallow as msh
|
||||
|
||||
from keyosk import datatypes
|
||||
from keyosk import fields as custom_fields
|
||||
|
||||
|
||||
def _default_sqlite_pragmas() -> Dict[str, Any]:
|
||||
"""Generate the default pragmas for the sqlite connection
|
||||
|
||||
Default are taken from
|
||||
`here <http://docs.peewee-orm.com/en/latest/peewee/database.html#recommended-settings>`_
|
||||
"""
|
||||
return {
|
||||
"journal_mode": "wal",
|
||||
"cache_size": -1 * 64000,
|
||||
"foreign_keys": 1,
|
||||
"ignore_check_constraints": 0,
|
||||
"synchronous": 0,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class KeyoskSQLiteStorageConfig:
|
||||
"""Config data container for the SQLite config options
|
||||
|
||||
:param path: Path to the SQLite database file
|
||||
:param pragmas: Mapping of SQLite pragmas to apply to the database connection
|
||||
"""
|
||||
|
||||
path: Path = Path("/usr/share/keyosk.db")
|
||||
pragmas: Mapping[str, Any] = field(default_factory=_default_sqlite_pragmas)
|
||||
|
||||
|
||||
class SQLiteStorageConfigSerializer(msh.Schema):
|
||||
"""De/serializer for the SQLite configuration parameters
|
||||
|
||||
Fields on this class map 1:1 with the dataclass parameters on the
|
||||
:class:`KeyoskSQLiteStorageConfig` class.
|
||||
"""
|
||||
|
||||
path = custom_fields.Path()
|
||||
pragmas = msh.fields.Dict(keys=msh.fields.String(), values=msh.fields.Raw())
|
||||
|
||||
@msh.post_load
|
||||
def _make_dataclass(self, data: Mapping[str, Any], *args, **kwargs):
|
||||
return KeyoskSQLiteStorageConfig(**data)
|
||||
|
||||
@msh.pre_dump
|
||||
def _unmake_dataclass(
|
||||
self, data: Union[Mapping[str, Any], KeyoskSQLiteStorageConfig], *args, **kwargs
|
||||
):
|
||||
if isinstance(data, KeyoskSQLiteStorageConfig):
|
||||
return asdict(data)
|
||||
return data
|
||||
|
||||
|
||||
@dataclass
|
||||
class KeyoskMariaStorageConfig:
|
||||
"""Config data container for the MariaDB config options
|
||||
|
||||
:param schema: Database schema to use
|
||||
:param host: IP address or hostname of the database server to connect to
|
||||
:param port: Port to connect to the database server on
|
||||
:param username: Username for connecting to the database server
|
||||
:param password: Password for the user account to use for connecting to the database
|
||||
server
|
||||
|
||||
.. note:: The MySQL driver treats the hosts ``localhost`` and ``127.0.0.1``
|
||||
differently: using ``localhost`` will cause the client to always attempt
|
||||
to use a socket connection, while ``127.0.0.1`` will cause the client to
|
||||
always attempt to use a TCP connection.
|
||||
"""
|
||||
|
||||
schema: str = "keyosk"
|
||||
host: str = "localhost"
|
||||
port: int = 3306
|
||||
username: str = "keyosk"
|
||||
password: Optional[str] = None
|
||||
|
||||
|
||||
class MariaStorageConfigSerializer(msh.Schema):
|
||||
"""De/serializer for the MariaDB configuration parameters
|
||||
|
||||
Fields on this class map 1:1 with the dataclass parameters on the
|
||||
:class:`KeyoskMariaStorageConfig` class.
|
||||
"""
|
||||
|
||||
schema = msh.fields.String()
|
||||
host = msh.fields.String()
|
||||
port = msh.fields.Integer(validate=msh.validate.Range(min=1, max=65535))
|
||||
username = msh.fields.String()
|
||||
password = msh.fields.String(allow_none=True)
|
||||
|
||||
@msh.post_load
|
||||
def _make_dataclass(self, data: Mapping[str, Any], *args, **kwargs):
|
||||
return KeyoskMariaStorageConfig(**data)
|
||||
|
||||
@msh.pre_dump
|
||||
def _unmake_dataclass(
|
||||
self, data: Union[Mapping[str, Any], KeyoskMariaStorageConfig], *args, **kwargs
|
||||
):
|
||||
if isinstance(data, KeyoskMariaStorageConfig):
|
||||
return asdict(data)
|
||||
return data
|
||||
|
||||
|
||||
@dataclass
|
||||
class KeyoskStorageConfig:
|
||||
"""Config data container for storage related parameters
|
||||
|
||||
:param backend: The backend database system the application should use
|
||||
:param sqlite: Configuration parameters for the SQLite backend
|
||||
:param maria: Configuration parameters for the MariaDB backend
|
||||
|
||||
.. note:: Only one of the ``sqlite`` or ``maria`` parameters will be used at any one
|
||||
time, depending on the value of the ``backend`` setting.
|
||||
"""
|
||||
|
||||
backend: datatypes.StorageBackend = datatypes.StorageBackend.SQLITE
|
||||
sqlite: KeyoskSQLiteStorageConfig = KeyoskSQLiteStorageConfig()
|
||||
maria: KeyoskMariaStorageConfig = KeyoskMariaStorageConfig()
|
||||
|
||||
|
||||
class StorageConfigSerializer(msh.Schema):
|
||||
"""De/serializer for the storage configuration parameters
|
||||
|
||||
Fields on this class map 1:1 with the dataclass parameters on the
|
||||
:class:`KeyoskStorageConfig` class.
|
||||
"""
|
||||
|
||||
backend = custom_fields.EnumItem(datatypes.StorageBackend, pretty_names=True)
|
||||
sqlite = msh.fields.Nested(SQLiteStorageConfigSerializer)
|
||||
maria = msh.fields.Nested(MariaStorageConfigSerializer)
|
||||
|
||||
@msh.post_load
|
||||
def _make_dataclass(self, data: Mapping[str, Any], *args, **kwargs):
|
||||
return KeyoskStorageConfig(**data)
|
||||
|
||||
@msh.pre_dump
|
||||
def _unmake_dataclass(
|
||||
self, data: Union[Mapping[str, Any], KeyoskStorageConfig], *args, **kwargs
|
||||
):
|
||||
if isinstance(data, KeyoskStorageConfig):
|
||||
return asdict(data)
|
||||
return data
|
82
keyosk/fields.py
Normal file
82
keyosk/fields.py
Normal file
@ -0,0 +1,82 @@
|
||||
"""Custom fields for handing de/serializing custom data types"""
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Type
|
||||
from typing import Union
|
||||
|
||||
import marshmallow as msh
|
||||
|
||||
|
||||
class EnumItem(msh.fields.Field):
|
||||
"""Translate between an enum and its value or name"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
enum: Type[Enum],
|
||||
*args,
|
||||
by_value: bool = False,
|
||||
pretty_names: bool = False,
|
||||
**kwargs,
|
||||
):
|
||||
"""Initialize the enum field
|
||||
:param enum: The base enumeration to use for de/serializing to/from. Passing in a name/value
|
||||
that does not appear in this enum during de/serialization will result in a
|
||||
:exc:`ValidationError` being raised.
|
||||
:param by_value: Whether to perform de/serialization using the enum name or enum value. By
|
||||
default the field will be de/serialized using the enum name, but passing
|
||||
this as true will perform de/serialization using the enum value.
|
||||
:param pretty_names: Whether to interperate the enum names as "pretty" names. This will
|
||||
convert between uppercase+underscore-delimited and
|
||||
lowercase+hyphen-delimited. For example, an enum named ``FOO_BAR_BAZ``
|
||||
would become ``foo-bar-baz``. This option has no effect if
|
||||
``by_value=True`` is passed.
|
||||
"""
|
||||
|
||||
super().__init__(*args, **kwargs)
|
||||
self._by_value = by_value
|
||||
self._pretty_names = pretty_names
|
||||
self.enum = enum
|
||||
|
||||
def _serialize(self, value: Enum, attr, obj, **kwargs) -> Any:
|
||||
"""Serialize an enumeration to either its name or value"""
|
||||
if getattr(self, "allow_none", False) is True and value is None:
|
||||
return None
|
||||
if self._by_value:
|
||||
return value.value
|
||||
if self._pretty_names:
|
||||
return self._to_pretty_name(value.name)
|
||||
return value.name
|
||||
|
||||
def _deserialize(self, value: Any, attr, data, **kwargs) -> Enum:
|
||||
"""Serialize the name or value of an enumeration to its corresponding enum"""
|
||||
try:
|
||||
if self._by_value:
|
||||
return self.enum(value)
|
||||
if self._pretty_names:
|
||||
if value in self.enum.__members__:
|
||||
raise KeyError(value) # Just gets us down to the keyerror handler
|
||||
return self.enum[self._from_pretty_name(value)]
|
||||
return self.enum[value]
|
||||
except ValueError as err:
|
||||
raise msh.ValidationError(err)
|
||||
except (KeyError, AttributeError) as err:
|
||||
raise msh.ValidationError(f"No {self.enum} named {err}")
|
||||
|
||||
@staticmethod
|
||||
def _to_pretty_name(value: str) -> str:
|
||||
return value.replace("_", "-").lower()
|
||||
|
||||
@staticmethod
|
||||
def _from_pretty_name(value: str) -> str:
|
||||
return value.replace("-", "_").upper()
|
||||
|
||||
|
||||
class Path(msh.fields.String):
|
||||
"""Translate between a string and a path object"""
|
||||
|
||||
def _serialize(self, value: Union[str, Path], *args, **kwargs) -> str:
|
||||
return super()._serialize(str(value), *args, **kwargs)
|
||||
|
||||
def _deserialize(self, value: str, *args, **kwargs) -> Path:
|
||||
return Path(super()._deserialize(value, *args, **kwargs))
|
Loading…
Reference in New Issue
Block a user