~launchpad-pqm/launchpad/devel

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
# Copyright 2009 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

"""Base classes for HTTP resources."""

__metaclass__ = type
__all__ = [
    'LibraryBackedByteStorage',
]


from cStringIO import StringIO

from lazr.restful.interfaces import IByteStorage
from zope.component import (
    getMultiAdapter,
    getUtility,
    )
from zope.interface import implements

from lp.services.webapp.interfaces import ICanonicalUrlData
from lp.services.librarian.interfaces import (
    ILibraryFileAliasSet,
    ILibraryFileAliasWithParent,
    )


class LibraryBackedByteStorage:
    """See `IByteStorage`."""
    implements(IByteStorage, ICanonicalUrlData)

    def __init__(self, entry, field):
        """Initialize as the backing storage for one entry's field."""
        self.entry = entry
        self.field = field
        self.file_alias = getattr(self.entry, self.field.__name__)

    @property
    def rootsite(self):
        """See `ICanonicalUrlData`"""
        return None

    @property
    def inside(self):
        """See `ICanonicalUrlData`"""
        return self.entry.context

    @property
    def path(self):
        """See `ICanonicalUrlData`"""
        return self.field.__name__

    @property
    def alias_url(self):
        """See `IByteStorage`."""
        if self.file_alias.restricted:
            lfa_with_parent = getMultiAdapter(
                (self.file_alias, self.entry.context),
                ILibraryFileAliasWithParent)
            token = lfa_with_parent.createToken()
            return self.file_alias.private_url + '?token=%s' % token
        return self.file_alias.getURL()

    @property
    def filename(self):
        """See `IByteStorage`."""
        if self.is_stored:
            return self.file_alias.filename
        return self.field.__name__

    @property
    def is_stored(self):
        """See `IByteStorage`."""
        return self.file_alias is not None

    def createStored(self, mediaType, representation, filename=None):
        """See `IByteStorage`."""
        if filename is None:
            filename = self.filename
        stored = getUtility(ILibraryFileAliasSet).create(
            name=filename, size=len(representation),
            file=StringIO(representation), contentType=mediaType)
        setattr(self.entry, self.field.__name__, stored)

    def deleteStored(self):
        """See `IByteStorage`."""
        setattr(self.entry, self.field.__name__, None)