/usr/lib/python3/dist-packages/radicale/storage/filesystem.py is in python3-radicale 1.1.1+20160115-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | # -*- coding: utf-8 -*-
#
# This file is part of Radicale Server - Calendar Server
# Copyright © 2012-2015 Guillaume Ayoub
#
# 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/>.
"""
Filesystem storage backend.
"""
import codecs
import os
import posixpath
import json
import time
import sys
from contextlib import contextmanager
from .. import config, ical, log, pathutils
FOLDER = os.path.expanduser(config.get("storage", "filesystem_folder"))
FILESYSTEM_ENCODING = sys.getfilesystemencoding()
try:
from dulwich.repo import Repo
GIT_REPOSITORY = Repo(FOLDER)
except:
GIT_REPOSITORY = None
# This function overrides the builtin ``open`` function for this module
# pylint: disable=W0622
@contextmanager
def open(path, mode="r"):
"""Open a file at ``path`` with encoding set in the configuration."""
# On enter
abs_path = os.path.join(FOLDER, path.replace("/", os.sep))
with codecs.open(abs_path, mode, config.get("encoding", "stock")) as fd:
yield fd
# On exit
if GIT_REPOSITORY and mode == "w":
path = os.path.relpath(abs_path, FOLDER)
GIT_REPOSITORY.stage([path])
committer = config.get("git", "committer")
GIT_REPOSITORY.do_commit(
path.encode("utf-8"), committer=committer.encode("utf-8"))
# pylint: enable=W0622
class Collection(ical.Collection):
"""Collection stored in a flat ical file."""
@property
def _filesystem_path(self):
"""Absolute path of the file at local ``path``."""
return pathutils.path_to_filesystem(self.path, FOLDER)
@property
def _props_path(self):
"""Absolute path of the file storing the collection properties."""
return self._filesystem_path + ".props"
def _create_dirs(self):
"""Create folder storing the collection if absent."""
if not os.path.exists(os.path.dirname(self._filesystem_path)):
os.makedirs(os.path.dirname(self._filesystem_path))
def save(self, text):
self._create_dirs()
with open(self._filesystem_path, "w") as fd:
fd.write(text)
def delete(self):
os.remove(self._filesystem_path)
os.remove(self._props_path)
@property
def text(self):
try:
with open(self._filesystem_path) as fd:
return fd.read()
except IOError:
return ""
@classmethod
def children(cls, path):
filesystem_path = pathutils.path_to_filesystem(path, FOLDER)
_, directories, files = next(os.walk(filesystem_path))
for filename in directories + files:
# make sure that the local filename can be translated
# into an internal path
if not pathutils.is_safe_path_component(filename):
log.LOGGER.debug("Skipping unsupported filename: %s", filename)
continue
rel_filename = posixpath.join(path, filename)
if cls.is_node(rel_filename) or cls.is_leaf(rel_filename):
yield cls(rel_filename)
@classmethod
def is_node(cls, path):
filesystem_path = pathutils.path_to_filesystem(path, FOLDER)
return os.path.isdir(filesystem_path)
@classmethod
def is_leaf(cls, path):
filesystem_path = pathutils.path_to_filesystem(path, FOLDER)
return (
os.path.isfile(filesystem_path) and not
filesystem_path.endswith(".props"))
@property
def last_modified(self):
modification_time = time.gmtime(
os.path.getmtime(self._filesystem_path))
return time.strftime("%a, %d %b %Y %H:%M:%S +0000", modification_time)
@property
@contextmanager
def props(self):
# On enter
properties = {}
if os.path.exists(self._props_path):
with open(self._props_path) as prop_file:
properties.update(json.load(prop_file))
old_properties = properties.copy()
yield properties
# On exit
self._create_dirs()
if old_properties != properties:
with open(self._props_path, "w") as prop_file:
json.dump(properties, prop_file)
|