ProPeace/adGroupSync
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

➕ replace tomlib with tomlkit

Browse source 
tomlkit can read and write toml files
This commit is contained in:
Marc Koch 2025-03-21 13:22:23 +01:00
parent 3effd5558f
commit 0d4d6d3f46
Signed by: marc.koch
GPG key ID: 12406554CFB028B9
8 changed files with 129 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/adgroupsync/__main__.py
View file

@ -5,7 +5,6 @@ from pathlib import Path
from ldap3 import SIMPLE
from ms_active_directory import ADDomain, ADGroup, ADUser, ADObject
from .exceptions import ScriptAlreadyRunningError
from .conf import (
AD_DOMAIN,
AD_USER_NAME,
@ -25,6 +24,7 @@ from .conf import (
NTFY_TOPIC,
NTFY_ACCESS_TOKEN,
)
from .exceptions import ScriptAlreadyRunningError
from .logger import setup_logging
from .models import RecentRun, CiviCrm, Ntfy

90
src/adgroupsync/conf.py
View file

@ -4,19 +4,83 @@ import os
from pathlib import Path
import pytz
import yaml
import tomlkit
from tomlkit.toml_file import TOMLFile
logger = logging.getLogger(__package__)
def create_config_file(dest: Path):
if dest.is_dir():
dest = dest / f"{__package__}_config.yml"
example_conf = Path(__file__).parent / 'resources' / 'example_config.yml'
with open(example_conf, "r") as source:
with open(dest, "w") as d:
d.writelines(source)
dest = dest / f"{__package__}_config.toml"
conf = tomlkit.document()
conf.add(tomlkit.comment(f"Configuration file for {__package__}"))
conf.add(tomlkit.nl())
# Add AD section
ad = tomlkit.table()
ad.add('DOMAIN', 'ad.example.com')
ad['DOMAIN'].comment('The domain of the Active Directory server')
ad.add('LDAP_SERVER', ['ldaps://server1.ad.example.com:636'])
ad['LDAP_SERVER'].comment('List of LDAP servers to connect to')
ad.add('PARENT_GROUP', 'Mailinglists')
ad['PARENT_GROUP'].comment('The parent group in Active Directory that '
'contains all groups that should be '
'synchronized')
ad.add('TIMEZONE', 'UTC')
ad.add('USER', tomlkit.string('example\\username', literal=True))
ad['USER'].comment('The username of the user to connect to the Active '
'Directory server. (use single quotes)')
ad.add('PASSWORD', 'xxxxxxxx')
ad['PASSWORD'].comment('The password of the user to connect to the Active '
'Directory')
conf.add('AD', ad)
# Add CiviCRM section
civicrm = tomlkit.table()
civicrm.add('BASE_URL', 'https://civicrm.example.com')
civicrm.add('API_KEY', 'xxxxxxxx')
civicrm['API_KEY'].comment('The API key of the CiviCRM user')
civicrm['BASE_URL'].comment('The URL of the CiviCRM server')
civicrm.add('BATCH_SIZE', 50)
civicrm['BATCH_SIZE'].comment('The batch size for the API requests to the '
'CiviCRM server - only applied to contact '
'sync (default: 50)')
civicrm.add('RETRIES', 3)
civicrm['RETRIES'].comment('The number of retries for the API requests to '
'the CiviCRM (default: 3)')
civicrm.add('IGNORE_SSL', False)
civicrm.value.item('IGNORE_SSL').comment('Allow insecure connections to '
'the CiviCRM server (default: '
'false)')
conf.add('CIVICRM', civicrm)
# Add Logging section
logging_ = tomlkit.table()
logging_.add('STDOUT_LOG_LEVEL', 'info')
logging_['STDOUT_LOG_LEVEL'].comment('The log level for the stdout logger '
'(default: info)')
logging_.add('FILE_LOG_LEVEL', 'info')
logging_['FILE_LOG_LEVEL'].comment('The log level for the file logger '
'(default: info)')
logging_.add('LOG_DIR', '/var/log/adGroupSync/')
logging_['LOG_DIR'].comment('The directory to store the log file')
conf.add('LOGGING', logging_)
# Add Ntfy section
ntfy = tomlkit.table()
ntfy.add(tomlkit.comment('Optional section for ntfy configuration'))
ntfy.add('URL', 'https://ntfy.example.com')
ntfy['URL'].comment('The URL of the ntfy server')
ntfy.add('TOPIC', 'adGroupSync')
ntfy['TOPIC'].comment('The topic to post the message to')
ntfy.add('ACCESS_TOKEN', 'tk_xxxxxxxxxxxxxxxxxxx')
ntfy['ACCESS_TOKEN'].comment('The access token for the ntfy server')
conf.add('NTFY', ntfy)
# Write configuration file
TOMLFile(dest).write(conf)
# Assign environment variables or configuration file values
AD_DOMAIN = os.getenv('AD_DOMAIN')
@ -77,8 +141,7 @@ try:
exit(0)
# Load configuration file
with open(config_file, 'r') as file:
config = yaml.safe_load(file)
config = TOMLFile(config_file).read()
# Get values from configuration file
AD_DOMAIN = AD_DOMAIN or config['AD']['DOMAIN']
@ -98,13 +161,16 @@ try:
CIVICRM_BATCH_SIZE = CIVICRM_BATCH_SIZE \
or config['CIVICRM']['BATCH_SIZE']
CIVICRM_RETRIES = CIVICRM_RETRIES \
or config['CIVICRM'].get('RETRIES', 3)
or config['CIVICRM'].get('RETRIES', 3)
CIVICRM_IGNORE_SSL = CIVICRM_IGNORE_SSL \
or bool(config['CIVICRM'].get('IGNORE_SSL', False))
NTFY_URL = NTFY_URL or config['NTFY'].get('URL') if 'NTFY' in config else None
NTFY_TOPIC = NTFY_TOPIC or config['NTFY'].get('TOPIC') if 'NTFY' in config else None
NTFY_URL = NTFY_URL or config['NTFY'].get(
'URL') if 'NTFY' in config else None
NTFY_TOPIC = NTFY_TOPIC or config['NTFY'].get(
'TOPIC') if 'NTFY' in config else None
NTFY_ACCESS_TOKEN = NTFY_ACCESS_TOKEN \
or config['NTFY'].get('ACCESS_TOKEN') if 'NTFY' in config else None
or config['NTFY'].get(
'ACCESS_TOKEN') if 'NTFY' in config else None
# Check if some required values are missing
required = {

57
src/adgroupsync/models.py
View file

@ -1,16 +1,15 @@
import datetime
import json
import logging
import tomllib
from collections import deque
from datetime import datetime as dt, timezone
from pathlib import Path
import pytz
import tomli_w
from civifang import api
from httpx import post
from ms_active_directory import ADUser, ADGroup
from tomlkit.toml_file import TOMLFile
from .enums import Priority
from .exceptions import ScriptAlreadyRunningError
@ -56,38 +55,46 @@ class RecentRun:
:param is_running:
:return:
"""
rr = recent_run if recent_run else self._datetime
is_running = is_running if is_running is not None else self._already_running
new_data = {
'recent-run': rr,
'is-running': is_running,
}
# Return if no data is provided
if recent_run is None and is_running is None:
return
# Read the existing data
toml_file = TOMLFile(self._file_path)
toml = toml_file.read()
# Update the values
if recent_run:
toml['recent_run'] = recent_run
if is_running is not None:
toml['is-running'] = is_running
# Write the data to the file
with open(self._file_path, 'wb') as f:
tomli_w.dump(new_data, f)
toml_file.write(toml)
def _read_data_from_file(self):
"""
Read the recent run time from the file
:return:
"""
with open(self._file_path, 'rb') as f:
data = tomllib.load(f)
self._already_running = data.get('is-running', False)
recent_run = data.get('recent-run')
# Read the data from the file
toml_file = TOMLFile(self._file_path)
toml = toml_file.read()
# Set the datetime to the recent run time
if not recent_run:
return
if isinstance(recent_run, datetime.datetime):
self._datetime = recent_run
elif isinstance(recent_run, float):
self._datetime = dt.fromtimestamp(recent_run) \
.astimezone(self._timezone)
else:
raise ValueError(
f"Invalid recent_run '{recent_run}' in {self._file_path}.")
self._already_running = toml.get('is-running', False)
recent_run = toml.get('recent-run')
# Set the datetime to the recent run time
if not recent_run:
return
if isinstance(recent_run, datetime.datetime):
self._datetime = recent_run
elif isinstance(recent_run, float):
self._datetime = dt.fromtimestamp(recent_run) \
.astimezone(self._timezone)
else:
raise ValueError(
f"Invalid recent_run '{recent_run}' in {self._file_path}.")
@property
def datetime(self) -> dt | None:

25
src/adgroupsync/resources/example_config.yml
View file

@ -1,25 +0,0 @@
AD:
DOMAIN: ad.example.com
USER: example\username
PASSWORD: xxxxxxxx
LDAP_SERVER:
- ldaps://server1.ad.example.com:636
PARENT_GROUP: Mailinglists
TIMEZONE: UTC
LOGGING:
STDOUT_LOG_LEVEL: info
FILE_LOG_LEVEL: info
LOG_DIR: /var/log/adGroupSync/
CIVICRM:
BASE_URL: https://civicrm.example.com
API_KEY: xxxxxxxx
BATCH_SIZE: 50
RETRIES: 3
# IGNORE_SSL: yes
NTFY:
URL: https://ntfy.example.com
TOPIC: adGroupSync
ACCESS_TOKEN: tk_xxxxxxxxxxxxxxxxxxx