Initial commit

pocketbase.py

@@ -0,0 +1,58 @@
+import urllib.request
+import json
+from util import MultiPartForm
+
+def authenticate_user(user, pwd):
+    payload = json.dumps({ "identity": user, "password": pwd }).encode()
+    headers = { 'Content-Type': "application/json" }
+    req = urllib.request.Request(url='http://127.0.0.1:8090/api/collections/users/auth-with-password', headers=headers, data=payload, method='POST')
+    with urllib.request.urlopen(req) as f:
+        assert f.status == 200
+        res = f.read().decode()
+        return json.loads(res)["token"]
+
+
+def update_user_name(user_id, name, auth_token):
+    payload = json.dumps({"name": name}).encode()
+    headers = { 
+        'Content-Type': "application/json",
+        'Authorization': f'Bearer {auth_token}'
+    }
+    req = urllib.request.Request(url=f"http://127.0.0.1:8090/api/collections/users/records/{user_id}", headers=headers, data=payload, method="PATCH")
+    with urllib.request.urlopen(req) as f:
+        assert f.status == 200
+
+
+def set_user_avatar(user_id, file, auth_token):
+    form = MultiPartForm()
+    #form.add_field("name", "tototototo")
+    form.add_file("avatar", file.name, file)
+    payload = bytes(form)
+    req = urllib.request.Request(url=f"http://127.0.0.1:8090/api/collections/users/records/{user_id}", data=payload, method="PATCH")
+    req.add_header("Authorization", f"Bearer {auth_token}")
+    req.add_header('Content-type', form.get_content_type())
+    req.add_header('Content-length', len(payload))
+
+    #print(req.headers)
+    parts = payload.split(b"\r\n")
+
+    for i, data in enumerate(parts):
+        if i > 3:
+            print(f"{data[:10].hex(" ")}... ({str(len(data)//1024) + "kb" if len(data) > 1024 else str(len(data)) + "bytes"}.)")
+        else:
+            print(data)
+    
+    with urllib.request.urlopen(req) as f:
+        assert f.status == 200
+        print(f.read().decode())
+
+
+def main():
+    auth_token = authenticate_user("azerty", "12345678")
+    azerty_user_id = "f3b7xs6kjeazhaj"
+    #update_user_name(azerty_user_id, "new_name", auth_token)
+    set_user_avatar(azerty_user_id, open("test.jpg", "rb"), auth_token)
+
+
+if __name__ == "__main__":
+    main()

util.py

@@ -0,0 +1,114 @@
+# https://pymotw.com/3/urllib.request/#uploading-files
+
+import io
+import mimetypes
+from urllib import request
+import uuid
+
+
+class MultiPartForm:
+    """Accumulate the data to be used when posting a form."""
+
+    def __init__(self):
+        self.form_fields = []
+        self.files = []
+        # Use a large random byte string to separate
+        # parts of the MIME data.
+        self.boundary = uuid.uuid4().hex.encode('utf-8')
+        return
+
+    def get_content_type(self):
+        return 'multipart/form-data; boundary={}'.format(
+            self.boundary.decode('utf-8'))
+
+    def add_field(self, name, value):
+        """Add a simple field to the form data."""
+        self.form_fields.append((name, value))
+
+    def add_file(self, fieldname, filename, fileHandle,
+                 mimetype=None):
+        """Add a file to be uploaded."""
+        body = fileHandle.read()
+        if mimetype is None:
+            mimetype = (
+                mimetypes.guess_type(filename)[0] or
+                'application/octet-stream'
+            )
+        self.files.append((fieldname, filename, mimetype, body))
+        return
+
+    @staticmethod
+    def _form_data(name):
+        return ('Content-Disposition: form-data; '
+                'name="{}"\r\n').format(name).encode('utf-8')
+
+    @staticmethod
+    def _attached_file(name, filename):
+        return ('Content-Disposition: file; '
+                'name="{}"; filename="{}"\r\n').format(
+                    name, filename).encode('utf-8')
+
+    @staticmethod
+    def _content_type(ct):
+        return 'Content-Type: {}\r\n'.format(ct).encode('utf-8')
+
+    def __bytes__(self):
+        """Return a byte-string representing the form data,
+        including attached files.
+        """
+        buffer = io.BytesIO()
+        boundary = b'--' + self.boundary + b'\r\n'
+
+        # Add the form fields
+        for name, value in self.form_fields:
+            buffer.write(boundary)
+            buffer.write(self._form_data(name))
+            buffer.write(b'\r\n')
+            buffer.write(value.encode('utf-8'))
+            buffer.write(b'\r\n')
+
+        # Add the files to upload
+        for f_name, filename, f_content_type, body in self.files:
+            buffer.write(boundary)
+            buffer.write(self._attached_file(f_name, filename))
+            buffer.write(self._content_type(f_content_type))
+            buffer.write(b'\r\n')
+            buffer.write(body)
+            buffer.write(b'\r\n')
+
+        buffer.write(b'--' + self.boundary + b'--\r\n')
+        return buffer.getvalue()
+
+
+if __name__ == '__main__':
+    # Create the form with simple fields
+    form = MultiPartForm()
+    form.add_field('firstname', 'Doug')
+    form.add_field('lastname', 'Hellmann')
+
+    # Add a fake file
+    form.add_file(
+        'biography', 'bio.txt',
+        fileHandle=io.BytesIO(b'Python developer and blogger.'))
+
+    # Build the request, including the byte-string
+    # for the data to be posted.
+    data = bytes(form)
+    r = request.Request('http://localhost:8080/', data=data)
+    r.add_header(
+        'User-agent',
+        'PyMOTW (https://pymotw.com/)',
+    )
+    r.add_header('Content-type', form.get_content_type())
+    r.add_header('Content-length', len(data))
+
+    print()
+    print('OUTGOING DATA:')
+    for name, value in r.header_items():
+        print('{}: {}'.format(name, value))
+    print()
+    print(r.data.decode('utf-8'))
+
+    print()
+    print('SERVER RESPONSE:')
+    print(request.urlopen(r).read().decode('utf-8'))