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
|
# Copyright (c) 2015 SUSE Linux GmbH. All rights reserved.
#
# This file is part of kiwi.
#
# kiwi 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.
#
# kiwi 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 kiwi. If not, see <http://www.gnu.org/licenses/>
#
import os
import errno
import logging
from stat import ST_MODE
from typing import List
# project
from kiwi.command import Command
log = logging.getLogger('kiwi')
class DataSync:
"""
**Sync data from a source directory to a target directory**
"""
def __init__(self, source_dir: str, target_dir: str) -> None:
"""
Create a new DataSync instance and initialize
sync source and target
:param str source_dir: source directory path name
:param str target_dir: target directory path name
"""
self.source_dir = source_dir
self.target_dir = target_dir
def sync_data(
self, options: List[str] = [], exclude: List[str] = [],
force_trailing_slash: bool = False
) -> None:
"""
Sync data from source to target using the rsync protocol
:param list options: rsync options
:param list exclude: file patterns to exclude
:param bool force_trailing_slash: add '/' to source_dir if not present
A speciality of the rsync tool is that it behaves differently
if the given source_dir ends with a '/' or not. If it ends
with a slash the data structure below will be synced to the
target_dir. If it does not end with a slash the source_dir
and its contents are synced to the target_dir. For example
.. code:: bash
source
└── some_data
1. $ rsync -a source target
target
└── source
└── some_data
2. $ rsync -a source/ target
target
└── some_data
The parameter force_trailing_slash can be used to make
sure rsync behaves like shown in the second case. If
set to true a '/' is appended to the given source_dir
if not already present
"""
if force_trailing_slash and not self.source_dir.endswith(os.sep):
self.source_dir += os.sep
target_entry_permissions = None
exclude_options = []
rsync_options = []
if options:
rsync_options = options
if not self.target_supports_extended_attributes():
warn_me = False
if '--xattrs' in rsync_options:
rsync_options.remove('--xattrs')
warn_me = True
if '--acls' in rsync_options:
rsync_options.remove('--acls')
warn_me = True
if warn_me:
log.warning(
'Extended attributes not supported for target: %s',
self.target_dir
)
if exclude:
for item in exclude:
exclude_options.append('--exclude')
exclude_options.append(
'/' + item
)
if os.path.exists(self.target_dir):
target_entry_permissions = os.stat(self.target_dir)[ST_MODE]
Command.run(
['rsync'] + rsync_options + exclude_options + [
self.source_dir, self.target_dir
]
)
if target_entry_permissions:
# rsync applies the permissions of the source directory
# also to the target directory which is unwanted because
# only permissions of the files and directories from the
# source directory and its contents should be transfered
# but not from the source directory itself. Therefore
# the permission bits of the target directory before the
# sync are applied back after sync to ensure they have
# not changed
os.chmod(self.target_dir, target_entry_permissions)
def target_supports_extended_attributes(self) -> bool:
"""
Check if the target directory supports extended filesystem
attributes
:return: True or False
:rtype: bool
"""
try:
os.getxattr(self.target_dir, 'user.mime_type')
except OSError as e:
log.debug(
f'Check for extended attributes on {self.target_dir} said: {e}'
)
if e.errno == errno.ENOTSUP:
return False
return True
|