#
# repos.py: public Python interface for repos components
#
# Subversion is a tool for revision control.
# See http://subversion.apache.org for more information.
#
######################################################################
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
######################################################################
from libsvn.repos import *
from svn.core import _unprefix_names, Pool
_unprefix_names(locals(), 'svn_repos_')
_unprefix_names(locals(), 'SVN_REPOS_')
__all__ = filter(lambda x: x.lower().startswith('svn_'), locals().keys())
del _unprefix_names
# Names that are not to be exported
import svn.core as _svncore, svn.fs as _svnfs, svn.delta as _svndelta
# Available change actions
CHANGE_ACTION_MODIFY = 0
CHANGE_ACTION_ADD = 1
CHANGE_ACTION_DELETE = 2
CHANGE_ACTION_REPLACE = 3
class ChangedPath:
__slots__ = [ 'item_kind', 'prop_changes', 'text_changed',
'base_path', 'base_rev', 'path', 'added', 'action',
]
def __init__(self,
item_kind, prop_changes, text_changed, base_path, base_rev,
path, added, action=None):
self.item_kind = item_kind
self.prop_changes = prop_changes
self.text_changed = text_changed
self.base_path = base_path
self.base_rev = base_rev
self.path = path
if action not in [None, CHANGE_ACTION_MODIFY, CHANGE_ACTION_ADD,
CHANGE_ACTION_DELETE, CHANGE_ACTION_REPLACE]:
raise Exception("unsupported change type")
self.action = action
### it would be nice to avoid this flag. however, without it, it would
### be quite difficult to distinguish between a change to the previous
### revision (which has a base_path/base_rev) and a copy from some
### other path/rev. a change in path is obviously add-with-history,
### but the same path could be a change to the previous rev or a restore
### of an older version. when it is "change to previous", I'm not sure
### if the rev is always repos.rev - 1, or whether it represents the
### created or time-of-checkout rev. so... we use a flag (for now)
### Note: This flag is also set for replaced paths unlike self.action
### which is either set to CHANGE_ACTION_ADD or CHANGE_ACTION_REPLACE
self.added = added
class ChangeCollector(_svndelta.Editor):
"""An editor that, when driven, walks a revision or a transaction and
incrementally invokes a callback with ChangedPath instances corresponding to
paths changed in that revision.
Available Since: 1.2.0
"""
# BATON FORMAT: [path, base_path, base_rev]
def __init__(self, fs_ptr, root, pool=None, notify_cb=None):
"""Construct a walker over the svn_fs_root_t ROOT, which must
be in the svn_fs_t FS_PTR. Invoke NOTIFY_CB with a single argument
of type ChangedPath for each change under ROOT.
At this time, two ChangedPath objects will be passed for a path that had
been replaced in the revision/transaction. This may change in the future.
### Can't we deduce FS_PTR from ROOT?
### POOL is unused
"""
self.fs_ptr = fs_ptr
self.changes = { } # path -> ChangedPathEntry()
self.roots = { } # revision -> svn_svnfs_root_t
self.notify_cb = notify_cb
self.props = { }
self.fs_root = root
# Figger out the base revision and root properties.
if _svnfs.is_revision_root(self.fs_root):
rev = _svnfs.revision_root_revision(self.fs_root)
self.base_rev = rev - 1
self.props = _svnfs.revision_proplist(self.fs_ptr, rev)
else:
txn_name = _svnfs.txn_root_name(self.fs_root)
txn_t = _svnfs.open_txn(self.fs_ptr, txn_name)
self.base_rev = _svnfs.txn_base_revision(txn_t)
self.props = _svnfs.txn_proplist(txn_t)
def get_root_props(self):
return self.props
def get_changes(self):
return self.changes
def _send_change(self, path):
if self.notify_cb:
change = self.changes.get(path)
if change:
self.notify_cb(change)
def _make_base_path(self, parent_path, path):
idx = path.rfind('/')
if parent_path:
parent_path = parent_path + '/'
if idx == -1:
return parent_path + path
return parent_path + path[idx+1:]
def _get_root(self, rev):
try:
return self.roots[rev]
except KeyError:
pass
root = self.roots[rev] = _svnfs.revision_root(self.fs_ptr, rev)
return root
def open_root(self, base_revision, dir_pool=None):
return ('', '', self.base_rev) # dir_baton
def delete_entry(self, path, revision, parent_baton, pool=None):
base_path = self._make_base_path(parent_baton[1], path)
if _svnfs.is_dir(self._get_root(parent_baton[2]), base_path):
item_type = _svncore.svn_node_dir
else:
item_type = _svncore.svn_node_file
self.changes[path] = ChangedPath(item_type,
False,
False,
base_path, # base_path
parent_baton[2], # base_rev
path, # path
False, # added
CHANGE_ACTION_DELETE,
)
self._send_change(path)
def add_directory(self, path, parent_baton,
copyfrom_path, copyfrom_revision, dir_pool=None):
action = path in self.changes and CHANGE_ACTION_REPLACE \
or CHANGE_ACTION_ADD
self.changes[path] = ChangedPath(_svncore.svn_node_dir,
False,
False,
copyfrom_path, # base_path
copyfrom_revision, # base_rev
path, # path
True, # added
action,
)
if copyfrom_path and (copyfrom_revision != -1):
base_path = copyfrom_path
else:
base_path = path
base_rev = copyfrom_revision
return (path, base_path, base_rev) # dir_baton
def open_directory(self, path, parent_baton, base_revision, dir_pool=None):
base_path = self._make_base_path(parent_baton[1], path)
return (path, base_path, parent_baton[2]) # dir_baton
def change_dir_prop(self, dir_baton, name, value, pool=None):
dir_path = dir_baton[0]
if dir_path in self.changes:
self.changes[dir_path].prop_changes = True
else:
# can't be added or deleted, so this must be CHANGED
self.changes[dir_path] = ChangedPath(_svncore.svn_node_dir,
True,
False,
dir_baton[1], # base_path
dir_baton[2], # base_rev
dir_path, # path
False, # added
CHANGE_ACTION_MODIFY,
)
def add_file(self, path, parent_baton,
copyfrom_path, copyfrom_revision, file_pool=None):
action = path in self.changes and CHANGE_ACTION_REPLACE \
or CHANGE_ACTION_ADD
self.changes[path] = ChangedPath(_svncore.svn_node_file,
False,
False,
copyfrom_path, # base_path
copyfrom_revision, # base_rev
path, # path
True, # added
action,
)
if copyfrom_path and (copyfrom_revision != -1):
base_path = copyfrom_path
else:
base_path = path
base_rev = copyfrom_revision
return (path, base_path, base_rev) # file_baton
def open_file(self, path, parent_baton, base_revision, file_pool=None):
base_path = self._make_base_path(parent_baton[1], path)
return (path, base_path, parent_baton[2]) # file_baton
def apply_textdelta(self, file_baton, base_checksum):
file_path = file_baton[0]
if file_path in self.changes:
self.changes[file_path].text_changed = True
else:
# an add would have inserted a change record already, and it can't
# be a delete with a text delta, so this must be a normal change.
self.changes[file_path] = ChangedPath(_svncore.svn_node_file,
False,
True,
file_baton[1], # base_path
file_baton[2], # base_rev
file_path, # path
False, # added
CHANGE_ACTION_MODIFY,
)
# no handler
return None
def change_file_prop(self, file_baton, name, value, pool=None):
file_path = file_baton[0]
if file_path in self.changes:
self.changes[file_path].prop_changes = True
else:
# an add would have inserted a change record already, and it can't
# be a delete with a prop change, so this must be a normal change.
self.changes[file_path] = ChangedPath(_svncore.svn_node_file,
True,
False,
file_baton[1], # base_path
file_baton[2], # base_rev
file_path, # path
False, # added
CHANGE_ACTION_MODIFY,
)
def close_directory(self, dir_baton):
self._send_change(dir_baton[0])
def close_file(self, file_baton, text_checksum):
self._send_change(file_baton[0])
class RevisionChangeCollector(ChangeCollector):
"""Deprecated: Use ChangeCollector.
This is a compatibility wrapper providing the interface of the
Subversion 1.1.x and earlier bindings.
Important difference: base_path members have a leading '/' character in
this interface."""
def __init__(self, fs_ptr, root, pool=None, notify_cb=None):
root = _svnfs.revision_root(fs_ptr, root)
ChangeCollector.__init__(self, fs_ptr, root, pool, notify_cb)
def _make_base_path(self, parent_path, path):
idx = path.rfind('/')
if idx == -1:
return parent_path + '/' + path
return parent_path + path[idx:]