ref: 0dbb5248276e2ed6af3f5be47d0ceda06bf31a69 paste.sr.ht/pastesrht/blueprints/api/pastes.py -rw-r--r-- 4.2 KiB View raw
                                                                                
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
from flask import Blueprint, abort, request
from hashlib import sha1
from pastesrht.access import has_access, UserAccess
from pastesrht.types import Paste, Blob, PasteFile, PasteVisibility
from pastesrht.webhooks import PasteWebhook
from srht.api import paginated_response
from srht.database import db
from srht.oauth import oauth, current_token
from srht.validation import Validation

pastes = Blueprint("api.pastes", __name__)

@pastes.route("/api/pastes")
@oauth("pastes:read")
def pastes_GET():
    return paginated_response(Paste.sha, Paste.query.filter(Paste.sha != None)
            .filter(Paste.user_id == current_token.user_id))

@pastes.route("/api/pastes", methods=["POST"])
@oauth("pastes:write")
def pastes_POST():
    valid = Validation(request)
    files = valid.require("files", cls=list)
    if not valid.ok:
        return valid.response
    paste = Paste()
    paste.user = current_token.user
    paste.user_id = current_token.user_id
    visibility = valid.optional("visibility",
            cls=PasteVisibility,
            default=PasteVisibility.public)
    paste.visibility = visibility
    db.session.flush()
    paste_sha = sha1()
    filenames = set()
    for i, f in enumerate(files):
        valid.expect(isinstance(f, dict),
                f"Expected files[{i}] to be dict", field=f"files[{i}]")
        contents = f.get("contents")
        contents = contents.replace('\r\n', '\n').replace('\r', '\n')
        filename = f.get("filename")
        valid.expect(contents, f"files[{i}].contents is required",
                field=f"files[{i}]")
        valid.expect(not contents or isinstance(contents, str),
                f"Expceted files[{i}].contents to be a string",
                field=f"files[{i}].contents")
        valid.expect(not filename or isinstance(filename, str),
                f"Expceted files[{i}].filename to be a string",
                field=f"files[{i}].filename")
        valid.expect(filename not in filenames,
                f"Filename {filename or 'none'} cannot be specified more "
                "than once", field=f"files[{i}].filename")
        if not valid.ok:
            continue
        filenames.update({filename})
        sha = sha1()
        sha.update(contents.encode())
        sha = sha.hexdigest()
        blob = Blob.query.filter(Blob.sha == sha).one_or_none()
        if not blob:
            blob = Blob()
            blob.sha = sha
            blob.contents = contents
            db.session.add(blob)
            db.session.flush()
        paste_file = PasteFile()
        paste_file.paste_id = paste.id
        paste_file.blob_id = blob.id
        paste_file.filename = filename
        paste_sha.update((paste_file.filename
            if paste_file.filename else "\0").encode())
        paste_sha.update(blob.sha.encode())
        db.session.add(paste_file)
    if not valid.ok:
        return valid.response
    paste_sha.update(str(current_token.user_id).encode())
    paste.sha = paste_sha.hexdigest()
    PasteWebhook.deliver(PasteWebhook.Events.paste_create,
            paste.to_dict(),
            PasteWebhook.Subscription.user_id == paste.user_id)
    db.session.commit()
    return paste.to_dict(), 201

@pastes.route("/api/pastes/<sha>")
@oauth("pastes:read")
def pastes_by_id_GET(sha):
    paste = Paste.query.filter(Paste.sha == sha).one_or_none()
    if not paste:
        abort(404)
    if not has_access(current_token.user, paste, UserAccess.read):
        abort(401)
    return paste.to_dict()

@pastes.route("/api/pastes/<sha>", methods=['PUT'])
@oauth("pastes:write")
def paste_by_id_PUT(sha):
    paste = Paste.query.filter(Paste.sha == sha).one_or_none()
    if not paste:
        abort(404)
    if not has_access(current_token.user, paste, UserAccess.write):
        abort(401)
    valid = Validation(request)
    visibility = valid.optional("visibility",
            cls=PasteVisibility,
            default=paste.visibility)
    paste.visibility = visibility
    db.session.commit()
    return paste.to_dict()

@pastes.route("/api/blobs/<sha>")
@oauth("pastes:read")
def blob_by_id_GET(sha):
    blob = Blob.query.filter(Blob.sha == sha).one_or_none()
    if not blob:
        abort(404)
    return blob.to_dict()

PasteWebhook.api_routes(blueprint=pastes, prefix="/api/pastes")