[Zope3-checkins] CVS: Zope3/src/zope/app/fssync - fsbundle.py:1.1
main.py:1.1
Fred L. Drake, Jr.
fred at zope.com
Fri Sep 5 18:13:15 EDT 2003
Update of /cvs-repository/Zope3/src/zope/app/fssync
In directory cvs.zope.org:/tmp/cvs-serv10506/src/zope/app/fssync
Added Files:
fsbundle.py main.py
Log Message:
- move the bundle tool to "zope.app.fssync" since it only makes sense
for the app server
- use the "zope.fssync.copier" module to handle object copying
- snapshot some prelimary code for an "unpack" operation, which will copy
objects out of a bundle into a site management folder (completely untested)
=== Added File Zope3/src/zope/app/fssync/fsbundle.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""High-level class to support bundle management on an fssync checkout.
$Id: fsbundle.py,v 1.1 2003/09/05 21:13:15 fdrake Exp $
"""
import os
import posixpath
import shutil
from zope.fssync.copier import ObjectCopier
from zope.fssync.fssync import FSSync
from zope.fssync.fsutil import Error
from zope.fssync.metadata import Metadata
BUNDLE_TYPE = "zope.app.services.bundle.Bundle"
FOLDER_TYPE = "zope.app.services.folder.SiteManagementFolder"
class FSBundle(object):
def __init__(self):
self.metadata = Metadata()
self.sync = FSSync(metadata=self.metadata)
# bundle operations
def create(self, path, type, factory, source=None):
if os.path.exists(path):
raise Error("%r already exists", path)
dir, name = os.path.split(path)
self.check_name(name)
self.check_directory(dir)
self.check_directory_known(dir)
if source is not None:
self.check_source(source, BUNDLE_TYPE, FOLDER_TYPE)
if type is None and factory is None:
srctype, srcfactory = self.metadata.gettypeinfo(source)
if srctype == FOLDER_TYPE:
factory = type = BUNDLE_TYPE
else:
# source is already a bundle; create the same type
type = srctype
factory = srcfactory
elif factory is None and type is None:
factory = type = BUNDLE_TYPE
if source is None:
self.sync.mkdir(path)
else:
copier = ObjectCopier(self.sync)
copier.copy(source, path, children=True)
self.settypeinfo(path, type, factory)
def unpack(self, source, target):
# source identifies the bundle to unpack
# target identifies a location to unpack to
if os.path.exists(target):
target_dir = target
# compute target name from prefix of source
pass
target = os.path.join(target_dir, target_name)
else:
target_dir, target_name = os.path.split(target)
self.check_source(source, BUNDLE_TYPE)
self.check_directory_known(target_dir)
# ...
self.settypeinfo(target, FOLDER_TYPE, FOLDER_TYPE)
# helper methods
def settypeinfo(self, path, type, factory):
entry = self.metadata.getentry(path)
assert entry.get("flag") == "added"
# override any existing type and factory
entry["factory"] = factory
entry["type"] = type
self.metadata.flush()
def check_source(self, source, *allowed_types):
# make sure the source is a site-management folder or a bundle
if not os.path.exists(source):
raise Error("%r does not exist", source)
if not os.path.isdir(source):
raise Error("%r must be a directory", source)
self.check_directory_known(os.path.dirname(source))
self.check_directory_known(source)
type, factory = self.metadata.gettypeinfo(source)
if type not in allowed_types:
if type == BUNDLE_TYPE:
pass
elif type == FOLDER_TYPE:
pass
else:
# don't know; play it safe
raise Error(
"%r doesn't appear to be a bundle or site-management folder",
source)
def check_directory(self, dir):
if dir:
if not os.path.exists(dir):
raise Error("%r does not exist", dir)
if not os.path.isdir(dir):
raise Error("%r is not a directory", dir)
# else: os.curdir assumed
def check_directory_known(self, dir):
dir = dir or os.curdir
entry = self.metadata.getentry(dir)
if not entry:
raise Error("nothing known about", dir)
def check_name(self, name):
if name.count("-") != 1:
raise Error("%r is not a legal bundle name", name)
basename, version = name.split("-")
self.check_version(version)
def check_version(self, s):
self.parse_version(s)
def parse_version(self, s):
parts = s.split(".")
if len(parts) not in (3, 4):
raise Error("%r is not a valid bundle version", s)
try:
n0 = int(parts[0])
n1 = int(parts[1])
n2 = int(parts[2])
except ValueError:
raise Error("%r is not a valid bundle version", s)
try:
p3 = int(parts[3])
except IndexError:
p3 = None
except ValueError:
p3 = parts[3]
return (n0, n1, n2, p3)
=== Added File Zope3/src/zope/app/fssync/main.py ===
##############################################################################
#
# Copyright (c) 2003 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Zope 3 bundle management utility.
Command line syntax summary:
%(program)s create BUNDLE SOURCE
``%(program)s help'' prints the global help (this message)
``%(program)s help command'' prints the local help for the command
"""
"""
$Id: main.py,v 1.1 2003/09/05 21:13:15 fdrake Exp $
"""
from zope.fssync.command import Command, Usage
from zope.app.fssync.fsbundle import FSBundle
def main():
"""Main program.
The return value is the suggested sys.exit() status code:
0 or None for success
2 for command line syntax errors
1 or other for later errors
"""
cmd = Command(usage=__doc__)
for func, aliases, short, long in command_table:
cmd.addCommand(func.__name__, func, short, long, aliases)
return cmd.main()
def create(opts, args):
"""%(program)s create BUNDLE SOURCE
Create a bundle from a site management folder or another bundle.
The bundle will only be created if the container is a site
management folder. BUNDLE must be a valid bundle name.
The contents of SOURCE are copied into the newly created bundle,
and are scheduled for addition to the database. The new bundle
can be manipulated using 'zsync add' and 'zsync revert' (and just
editing the contents) as needed before committing it to the
database.
"""
factory = None
type = None
for opt, arg in opts:
if opt in ("-f", "--factory"):
if factory:
raise Usage("-f/--factory can only be given once")
factory = arg
elif opt in ("-t", "--type"):
if type:
raise Usage("-t/--type can only be given once")
type = arg
source = None
if len(args) == 1:
path = args[0]
elif len(args) == 2:
path, source = args
else:
raise Usage("create requires exactly one path")
fs = FSBundle()
fs.create(path, type, factory, source)
def unpack(opts, args):
"""%(program)s unpack bundle [dest]
"""
if len(args) < 1:
raise Usage("unpack requires a bundle")
if len(args) > 2:
raise Usage("unpack allows at most two args")
source = args[0]
if len(args) == 1:
target = os.curdir
else:
target = args[1]
fs = FSBundle()
fs.unpack(source, target)
command_table = [
# name is taken from the function name
# function, aliases, short opts, long opts
(create, "", "f:t:", "factory= type="),
(unpack, "", "", ""),
]
More information about the Zope3-Checkins
mailing list