sharing: csv based config store
This commit is contained in:
Peter Bieringer
530
radicale/sharing/csv.py
Normal file
530
radicale/sharing/csv.py
Normal file
@@ -0,0 +1,530 @@
|
||||
# This file is part of Radicale Server - Calendar Server
|
||||
# Copyright © 2026-2026 Peter Bieringer <pb@bieringer.de>
|
||||
#
|
||||
# This library is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This library is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with Radicale. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import csv
|
||||
import logging
|
||||
import os
|
||||
from typing import Union
|
||||
|
||||
from radicale import config, sharing
|
||||
from radicale.log import logger
|
||||
|
||||
""" CVS based sharing by token or map """
|
||||
|
||||
|
||||
class Sharing(sharing.BaseSharing):
|
||||
_lines: int = 0
|
||||
_sharing_cache: list[dict] = []
|
||||
_sharing_db_file: str
|
||||
|
||||
# Overloaded functions
|
||||
def init_database(self) -> bool:
|
||||
logger.debug("sharing database initialization for type 'csv'")
|
||||
sharing_db_file = self.configuration.get("sharing", "database_path")
|
||||
if sharing_db_file == "":
|
||||
folder = self.configuration.get("storage", "filesystem_folder")
|
||||
folder_db = os.path.join(folder, "collection-db")
|
||||
sharing_db_file = os.path.join(folder_db, "sharing.csv")
|
||||
logger.info("sharing database filename not provided, use default: %r", sharing_db_file)
|
||||
else:
|
||||
logger.info("sharing database filename: %r", sharing_db_file)
|
||||
|
||||
if not os.path.exists(folder_db):
|
||||
logger.warning("sharing database folder is not existing: %r (create now)", folder_db)
|
||||
try:
|
||||
os.mkdir(folder_db)
|
||||
except Exception as e:
|
||||
logger.error("sharing database folder cannot be created (check permissions): %r (%r)", folder_db, e)
|
||||
return False
|
||||
logger.info("sharing database folder successfully created: %r", folder_db)
|
||||
|
||||
if not os.path.exists(sharing_db_file):
|
||||
logger.warning("sharing database is not existing: %r", sharing_db_file)
|
||||
try:
|
||||
if self._create_empty_csv(sharing_db_file) is not True:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("sharing database (empty) cannot be created (check permissions): %r (%r)", sharing_db_file, e)
|
||||
return False
|
||||
logger.info("sharing database (empty) successfully created: %r", sharing_db_file)
|
||||
else:
|
||||
logger.info("sharing database exists: %r", sharing_db_file)
|
||||
|
||||
# read database
|
||||
try:
|
||||
if self._load_csv(sharing_db_file) is not True:
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error("sharing database load failed: %r (%r)", sharing_db_file, e)
|
||||
return False
|
||||
logger.info("sharing database load successful: %r (lines=%d)", sharing_db_file, self._lines)
|
||||
self._sharing_db_file = sharing_db_file
|
||||
return True
|
||||
|
||||
def get_database_info(self) -> Union[dict, None]:
|
||||
database_info = {'type': "csv"}
|
||||
return database_info
|
||||
|
||||
def verify_database(self) -> bool:
|
||||
logger.info("sharing database (csv) verification begin")
|
||||
logger.info("sharing database (csv) file: %r", self._sharing_db_file)
|
||||
logger.info("sharing database (csv) loaded entries: %d", self._lines)
|
||||
# nothing more todo for CSV
|
||||
logger.info("sharing database (csv) verification end")
|
||||
return True
|
||||
|
||||
def get_sharing(self,
|
||||
ShareType: str,
|
||||
PathOrToken: str,
|
||||
User: Union[str, None] = None) -> Union[dict, None]:
|
||||
""" retrieve sharing target and attributes by map """
|
||||
# Lookup
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing: lookup ShareType=%r PathOrToken=%r User=%r)", ShareType, PathOrToken, User)
|
||||
|
||||
index = 0
|
||||
found = False
|
||||
for row in self._sharing_cache:
|
||||
if index == 0:
|
||||
# skip fieldnames
|
||||
pass
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing: check row: %r", row)
|
||||
if row['ShareType'] != ShareType:
|
||||
pass
|
||||
elif row['PathOrToken'] != PathOrToken:
|
||||
pass
|
||||
elif User is not None and row['User'] != User:
|
||||
pass
|
||||
elif row['EnabledByOwner'] is not True:
|
||||
pass
|
||||
elif row['ShareType'] == "map":
|
||||
if row['EnabledByUser'] is not True:
|
||||
pass
|
||||
else:
|
||||
found = True
|
||||
break
|
||||
else:
|
||||
found = True
|
||||
break
|
||||
index += 1
|
||||
|
||||
if found:
|
||||
PathMapped = row['PathMapped']
|
||||
Owner = row['Owner']
|
||||
UserShare = row['User']
|
||||
Permissions = row['Permissions']
|
||||
Hidden: bool = (row['HiddenByOwner'] or row['HiddenByUser'])
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing: map %r to %r (Owner=%r User=%r Permissions=%r Hidden=%s)", PathOrToken, PathMapped, Owner, UserShare, Permissions, Hidden)
|
||||
return {
|
||||
"mapped": True,
|
||||
"PathOrToken": PathOrToken,
|
||||
"PathMapped": PathMapped,
|
||||
"Owner": Owner,
|
||||
"User": UserShare,
|
||||
"Hidden": Hidden,
|
||||
"Permissions": Permissions}
|
||||
return None
|
||||
|
||||
def list_sharing(self,
|
||||
OwnerOrUser: Union[str, None] = None,
|
||||
ShareType: Union[str, None] = None,
|
||||
PathOrToken: Union[str, None] = None,
|
||||
PathMapped: Union[str, None] = None,
|
||||
User: Union[str, None] = None,
|
||||
EnabledByOwner: Union[bool, None] = None,
|
||||
EnabledByUser: Union[bool, None] = None,
|
||||
HiddenByOwner: Union[bool, None] = None,
|
||||
HiddenByUser: Union[bool, None] = None) -> list[dict]:
|
||||
""" retrieve sharing """
|
||||
row: dict
|
||||
index = 0
|
||||
result = []
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/list/called: ShareType=%r OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r HiddenByOwner=%s HiddenByUser=%s", ShareType, OwnerOrUser, User, PathOrToken, PathMapped, HiddenByOwner, HiddenByUser)
|
||||
|
||||
for row in self._sharing_cache:
|
||||
if index == 0:
|
||||
# skip fieldnames
|
||||
pass
|
||||
else:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/list/row: test: %r", row)
|
||||
if ShareType is not None and row['ShareType'] != ShareType:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/list/row: skip by ShareType")
|
||||
pass
|
||||
elif OwnerOrUser is not None and (row['Owner'] != OwnerOrUser and row['User'] != OwnerOrUser):
|
||||
pass
|
||||
elif User is not None and row['User'] != User:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/list/row: skip by User")
|
||||
pass
|
||||
elif PathOrToken is not None and row['PathOrToken'] != PathOrToken:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/list/row: skip by PathOrToken")
|
||||
pass
|
||||
elif PathMapped is not None and row['PathMapped'] != PathMapped:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/list/row: skip by PathMapped")
|
||||
pass
|
||||
elif EnabledByOwner is not None and row['EnabledByOwner'] != EnabledByOwner:
|
||||
pass
|
||||
elif EnabledByUser is not None and row['EnabledByUser'] != EnabledByUser:
|
||||
pass
|
||||
elif HiddenByOwner is not None and row['HiddenByOwner'] != HiddenByOwner:
|
||||
pass
|
||||
elif HiddenByUser is not None and row['HiddenByUser'] != HiddenByUser:
|
||||
pass
|
||||
else:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/list/row: add: %r", row)
|
||||
result.append(row)
|
||||
index += 1
|
||||
return result
|
||||
|
||||
def create_sharing(self,
|
||||
ShareType: str,
|
||||
PathOrToken: str, PathMapped: str,
|
||||
Owner: str, User: str,
|
||||
Permissions: str = "r",
|
||||
EnabledByOwner: bool = False, EnabledByUser: bool = False,
|
||||
HiddenByOwner: bool = True, HiddenByUser: bool = True,
|
||||
Timestamp: int = 0) -> dict:
|
||||
""" create sharing """
|
||||
row: dict
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing: ShareType=%r", ShareType)
|
||||
if ShareType == "token":
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/token/create: PathOrToken=%r Owner=%r PathMapped=%r User=%r Permissions=%r", PathOrToken, Owner, PathMapped, User, Permissions)
|
||||
# check for duplicate token entry
|
||||
for row in self._sharing_cache:
|
||||
if row['ShareType'] != "token":
|
||||
continue
|
||||
if row['PathOrToken'] == PathOrToken:
|
||||
# must be unique systemwide
|
||||
logger.error("sharing/token/create: PathOrToken already exists: PathOrToken=%r", PathOrToken)
|
||||
return {"status": "conflict"}
|
||||
elif ShareType == "map":
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/map/create: PathOrToken=%r Owner=%r PathMapped=%r User=%r Permissions=%r", PathOrToken, Owner, PathMapped, User, Permissions)
|
||||
# check for duplicate map entry
|
||||
for row in self._sharing_cache:
|
||||
if row['ShareType'] != "map":
|
||||
continue
|
||||
if row['PathMapped'] == PathMapped and row['User'] == User and row['PathOrToken'] == PathOrToken:
|
||||
# must be unique systemwide
|
||||
logger.error("sharing/map/create: entry already exists: PathMapped=%r User=%r", PathMapped, User)
|
||||
return {"status": "conflict"}
|
||||
else:
|
||||
return {"status": "error"}
|
||||
|
||||
row = {"ShareType": ShareType,
|
||||
"PathOrToken": PathOrToken,
|
||||
"PathMapped": PathMapped,
|
||||
"Owner": Owner,
|
||||
"User": User,
|
||||
"Permissions": Permissions,
|
||||
"EnabledByOwner": EnabledByOwner,
|
||||
"EnabledByUser": EnabledByUser,
|
||||
"HiddenByOwner": HiddenByOwner,
|
||||
"HiddenByUser": HiddenByUser,
|
||||
"TimestampCreated": Timestamp,
|
||||
"TimestampUpdated": Timestamp}
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/*/create: add row: %r", row)
|
||||
self._sharing_cache.append(row)
|
||||
|
||||
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
|
||||
if self._write_csv(self._sharing_db_file):
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/create: write CSV done", ShareType)
|
||||
return {"status": "success"}
|
||||
logger.error("sharing/%s/create: cannot update CSV database", ShareType)
|
||||
return {"status": "error"}
|
||||
|
||||
def update_sharing(self,
|
||||
ShareType: str,
|
||||
PathOrToken: str,
|
||||
Owner: Union[str, None] = None,
|
||||
User: Union[str, None] = None,
|
||||
PathMapped: Union[str, None] = None,
|
||||
Permissions: Union[str, None] = None,
|
||||
EnabledByOwner: Union[bool, None] = None,
|
||||
HiddenByOwner: Union[bool, None] = None,
|
||||
Timestamp: int = 0) -> dict:
|
||||
""" update sharing """
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/update: PathOrToken=%r Owner=%r PathMapped=%r", ShareType, PathOrToken, Owner, PathMapped)
|
||||
|
||||
# lookup token
|
||||
found = False
|
||||
index = 0
|
||||
for row in self._sharing_cache:
|
||||
if index == 0:
|
||||
# skip fieldnames
|
||||
pass
|
||||
if row['ShareType'] != ShareType:
|
||||
pass
|
||||
elif row['PathOrToken'] != PathOrToken:
|
||||
pass
|
||||
else:
|
||||
found = True
|
||||
break
|
||||
index += 1
|
||||
|
||||
if found:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/update: found index=%d", ShareType, index)
|
||||
if Owner is not None and row['Owner'] != Owner:
|
||||
return {"status": "permission-denied"}
|
||||
if User is not None and row['User'] != User:
|
||||
return {"status": "permission-denied"}
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/update: Owner=%r PathOrToken=%r index=%d", ShareType, Owner, PathOrToken, index)
|
||||
logger.debug("TRACE/sharing/%s/update: orig row=%r", ShareType, row)
|
||||
|
||||
# CSV: remove+adjust+readd
|
||||
if PathMapped is not None:
|
||||
row["PathMapped"] = PathMapped
|
||||
if Permissions is not None:
|
||||
row["Permissions"] = Permissions
|
||||
if User is not None:
|
||||
row["User"] = User
|
||||
if EnabledByOwner is not None:
|
||||
row["EnabledByOwner"] = EnabledByOwner
|
||||
if HiddenByOwner is not None:
|
||||
row["HiddenByOwner"] = HiddenByOwner
|
||||
# update timestamp
|
||||
row["TimestampUpdated"] = Timestamp
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/update: adj row=%r", ShareType, row)
|
||||
|
||||
# replace row
|
||||
self._sharing_cache.pop(index)
|
||||
self._sharing_cache.append(row)
|
||||
|
||||
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
|
||||
if self._write_csv(self._sharing_db_file):
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/update: write CSV done", ShareType)
|
||||
return {"status": "success"}
|
||||
logger.error("sharing/%s/update: cannot update CSV database", ShareType)
|
||||
return {"status": "error"}
|
||||
else:
|
||||
return {"status": "not-found"}
|
||||
|
||||
def delete_sharing(self,
|
||||
ShareType: str,
|
||||
PathOrToken: str, Owner: str,
|
||||
PathMapped: Union[str, None] = None) -> dict:
|
||||
""" delete sharing """
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/delete: PathOrToken=%r Owner=%r PathMapped=%r", ShareType, PathOrToken, Owner, PathMapped)
|
||||
|
||||
# lookup token
|
||||
found = False
|
||||
index = 0
|
||||
for row in self._sharing_cache:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/delete: check: %r", ShareType, row)
|
||||
if index == 0:
|
||||
# skip fieldnames
|
||||
pass
|
||||
if row['ShareType'] != ShareType:
|
||||
pass
|
||||
elif row['PathOrToken'] != PathOrToken:
|
||||
pass
|
||||
else:
|
||||
if ShareType == "map":
|
||||
# extra filter
|
||||
if row['PathMapped'] != PathMapped:
|
||||
pass
|
||||
else:
|
||||
found = True
|
||||
break
|
||||
else:
|
||||
found = True
|
||||
break
|
||||
index += 1
|
||||
|
||||
if found:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/delete: found index=%d", ShareType, index)
|
||||
if row['Owner'] != Owner:
|
||||
return {"status": "permission-denied"}
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/delete: Owner=%r PathOrToken=%r index=%d", ShareType, Owner, PathOrToken, index)
|
||||
self._sharing_cache.pop(index)
|
||||
|
||||
with self._storage.acquire_lock("w", Owner, path=self._sharing_db_file):
|
||||
if self._write_csv(self._sharing_db_file):
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing_by_token: write CSV done")
|
||||
return {"status": "success"}
|
||||
logger.error("sharing/%s/delete: cannot update CSV database", ShareType)
|
||||
return {"status": "error"}
|
||||
else:
|
||||
return {"status": "not-found"}
|
||||
|
||||
def toggle_sharing(self,
|
||||
ShareType: str,
|
||||
PathOrToken: str,
|
||||
OwnerOrUser: str,
|
||||
Action: str,
|
||||
PathMapped: Union[str, None] = None,
|
||||
User: Union[str, None] = None,
|
||||
Timestamp: int = 0) -> dict:
|
||||
""" toggle sharing """
|
||||
row: dict
|
||||
|
||||
if Action not in sharing.API_SHARE_TOGGLES_V1:
|
||||
# should not happen
|
||||
raise
|
||||
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/%s: OwnerOrUser=%r User=%r PathOrToken=%r PathMapped=%r", ShareType, Action, OwnerOrUser, User, PathOrToken, PathMapped)
|
||||
|
||||
# lookup entry
|
||||
found = False
|
||||
index = 0
|
||||
for row in self._sharing_cache:
|
||||
if index == 0:
|
||||
# skip fieldnames
|
||||
pass
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/*/" + Action + ": check: %r", row)
|
||||
if row['ShareType'] != ShareType:
|
||||
pass
|
||||
elif row['PathOrToken'] != PathOrToken:
|
||||
pass
|
||||
elif PathMapped is not None and row['PathMapped'] != PathMapped:
|
||||
pass
|
||||
elif row['Owner'] == OwnerOrUser:
|
||||
found = True
|
||||
break
|
||||
else:
|
||||
found = True
|
||||
break
|
||||
index += 1
|
||||
|
||||
if found:
|
||||
# if logger.isEnabledFor(logging.DEBUG):
|
||||
# logger.debug("TRACE/sharing/*/" + Action + ": found: %r", row)
|
||||
if User is not None and row['User'] != User:
|
||||
return {"status": "permission-denied"}
|
||||
elif row['Owner'] == OwnerOrUser:
|
||||
pass
|
||||
elif row['User'] == OwnerOrUser:
|
||||
pass
|
||||
else:
|
||||
return {"status": "permission-denied"}
|
||||
|
||||
# TODO: locking
|
||||
if row['Owner'] == OwnerOrUser:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/%s: Owner=%r User=%r PathOrToken=%r index=%d", ShareType, Action, OwnerOrUser, User, PathOrToken, index)
|
||||
if Action == "disable":
|
||||
row['EnabledByOwner'] = False
|
||||
elif Action == "enable":
|
||||
row['EnabledByOwner'] = True
|
||||
elif Action == "hide":
|
||||
row['HiddenByOwner'] = True
|
||||
elif Action == "unhide":
|
||||
row['HiddenByOwner'] = False
|
||||
row['TimestampUpdated'] = Timestamp
|
||||
if row['User'] == OwnerOrUser:
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE/sharing/%s/%s: User=%r PathOrToken=%r index=%d", ShareType, Action, OwnerOrUser, PathOrToken, index)
|
||||
if Action == "disable":
|
||||
row['EnabledByUser'] = False
|
||||
elif Action == "enable":
|
||||
row['EnabledByUser'] = True
|
||||
elif Action == "hide":
|
||||
row['HiddenByUser'] = True
|
||||
elif Action == "unhide":
|
||||
row['HiddenByUser'] = False
|
||||
|
||||
row['TimestampUpdated'] = Timestamp
|
||||
|
||||
# remove
|
||||
self._sharing_cache.pop(index)
|
||||
# readd
|
||||
self._sharing_cache.append(row)
|
||||
|
||||
with self._storage.acquire_lock("w", OwnerOrUser, path=self._sharing_db_file):
|
||||
if self._write_csv(self._sharing_db_file):
|
||||
if logger.isEnabledFor(logging.DEBUG):
|
||||
logger.debug("TRACE: write CSV done")
|
||||
return {"status": "success"}
|
||||
logger.error("sharing: cannot update CSV database")
|
||||
return {"status": "error"}
|
||||
else:
|
||||
return {"status": "not-found"}
|
||||
|
||||
# local functions
|
||||
def _create_empty_csv(self, file: str) -> bool:
|
||||
with self._storage.acquire_lock("w", None, path=file):
|
||||
with open(file, 'w', newline='') as csvfile:
|
||||
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1)
|
||||
writer.writeheader()
|
||||
return True
|
||||
|
||||
def _load_csv(self, file: str) -> bool:
|
||||
logger.debug("sharing database load begin: %r", file)
|
||||
with self._storage.acquire_lock("r", None):
|
||||
with open(file, 'r', newline='') as csvfile:
|
||||
reader = csv.DictReader(csvfile, fieldnames=sharing.DB_FIELDS_V1)
|
||||
self._lines = 0
|
||||
for row in reader:
|
||||
# logger.debug("sharing database load read: %r", row)
|
||||
if self._lines == 0:
|
||||
# header line, check
|
||||
for fieldname in sharing.DB_FIELDS_V1:
|
||||
logger.debug("sharing database load check fieldname: %r", fieldname)
|
||||
if fieldname not in row:
|
||||
logger.debug("sharing database is incompatible: %r", file)
|
||||
return False
|
||||
# convert txt to bool
|
||||
if self._lines > 0:
|
||||
for fieldname in sharing.DB_FIELDS_V1_BOOL:
|
||||
row[fieldname] = config._convert_to_bool(row[fieldname])
|
||||
for fieldname in sharing.DB_FIELDS_V1_INT:
|
||||
row[fieldname] = int(row[fieldname])
|
||||
# check for duplicates
|
||||
dup = False
|
||||
for row_cached in self._sharing_cache:
|
||||
if row == row_cached:
|
||||
dup = True
|
||||
break
|
||||
if dup:
|
||||
continue
|
||||
# logger.debug("sharing database load add: %r", row)
|
||||
self._sharing_cache.append(row)
|
||||
self._lines += 1
|
||||
logger.debug("sharing database load end: %r", file)
|
||||
return True
|
||||
|
||||
def _write_csv(self, file: str) -> bool:
|
||||
with open(file, 'w', newline='') as csvfile:
|
||||
writer = csv.DictWriter(csvfile, fieldnames=sharing.DB_FIELDS_V1)
|
||||
writer.writerows(self._sharing_cache)
|
||||
return True
|
||||
Reference in New Issue
Block a user