feat: complete image-sharing appWSGI Tutorial

.gitignore

@@ -160,3 +160,5 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
 
+images/
+

README.md

@@ -7,3 +7,52 @@
 ```
 python3 -m pip install -r requirements.txt
 ```
+
+
+## WSGI Tutorial
+
+实现了 WSGI Tutorial 的 image-sharing app
+
+提交图片
+
+```shell
+http POST localhost:8000/images Content-Type:image/png < /Users/apple/IMG_0976_little.png
+```
+
+查询图片
+
+```shell
+http GET localhost:8000/images/<image_name.xxx>
+```
+
+
+## 运行测试
+
+
+### 运行单元测试 
+
+#### 运行所有测试
+
+```shell
+coverage run -m pytest tests
+```
+
+#### 运行单个测试
+
+```shell
+pytest tests -k <test_func_name>
+```
+
+### 运行功能测试
+
+#### 启动 app
+
+```shell
+LOOK_STORAGE_PATH=./images gunicorn --reload 'app.app:get_app()'
+```
+
+#### 运行测试
+
+```shell
+pytest tests -k test_posted_image_gets_saved
+```

app/app.py

@@ -1,7 +1,19 @@
+import os
+
 import falcon
 
-from .images import Resource
+from .images import Collection, Item, ImageStore
 
-app = application = falcon.App()
 
-app.add_route('/images', Resource())
+def create_app(image_store):
+    app = falcon.App()
+    app.add_route('/images', Collection(image_store))
+    app.add_route('/images/{name}', Item(image_store))
+    return app
+
+
+def get_app():
+    storage_path = os.environ.get('LOOK_STORAGE_PATH', './images')
+    image_store = ImageStore(storage_path)
+    return create_app(image_store)
+

app/images.py

@@ -1,10 +1,34 @@
+import io
+import os
+import re
+import uuid
+import mimetypes
+
 import falcon
 import msgpack
 
 
-class Resource:
+ALLOWED_IMAGE_TYPES = (
+    'image/gif',
+    'image/jpeg',
+    'image/png',
+)
+
+
+def validate_image_type(req, resp, resource, params):
+    if req.content_type not in ALLOWED_IMAGE_TYPES:
+        msg = 'Image type not allowed. Must be PNG, JPEG, or GIF'
+        raise falcon.HTTPBadRequest(title='Bad request', description=msg)
+
+
+class Collection:
+
+    def __init__(self, image_store):
+        self._image_store = image_store
 
     def on_get(self, req, resp):
+        # TODO: Modify this to return a list of href's based on
+        # what images are actually available.
         doc = {
             'images': [
                 {
@@ -16,3 +40,63 @@ class Resource:
         resp.data = msgpack.packb(doc, use_bin_type=True)
         resp.content_type = falcon.MEDIA_MSGPACK
         resp.status = falcon.HTTP_200
+
+    @falcon.before(validate_image_type)
+    def on_post(self, req, resp):
+        name = self._image_store.save(req.stream, req.content_type)
+        resp.status = falcon.HTTP_201
+        resp.location = '/images/' + name
+
+
+class Item:
+
+    def __init__(self, image_store):
+        self._image_store = image_store
+
+    def on_get(self, req, resp, name):
+        resp.content_type = mimetypes.guess_type(name)[0]
+
+        try:
+            resp.stream, resp.content_length = self._image_store.open(name)
+        except IOError:
+            # Normally you would also log the error.
+            raise falcon.HTTPNotFound()
+
+
+class ImageStore:
+
+    _CHUNK_SIZE_BYTES = 4096
+    _IMAGE_NAME_PATTERN = re.compile(
+        '[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\.[a-z]{2,4}$'
+    )
+
+    def __init__(self, storage_path, uuidgen=uuid.uuid4, fopen=io.open):
+        self._storage_path = storage_path
+        self._uuidgen = uuidgen
+        self._fopen = fopen
+
+    def save(self, image_stream, image_content_type):
+        ext = mimetypes.guess_extension(image_content_type)
+        name = '{uuid}{ext}'.format(uuid=self._uuidgen(), ext=ext)
+        image_path = os.path.join(self._storage_path, name)
+
+        with self._fopen(image_path, 'wb') as image_file:
+            while True:
+                chunk = image_stream.read(self._CHUNK_SIZE_BYTES)
+                if not chunk:
+                    break
+
+                image_file.write(chunk)
+
+        return name
+
+    def open(self, name):
+        # Always validate untrusted input!
+        if not self._IMAGE_NAME_PATTERN.match(name):
+            raise IOError('File not found')
+
+        image_path = os.path.join(self._storage_path, name)
+        stream = self._fopen(image_path, 'rb')
+        content_length = os.path.getsize(image_path)
+
+        return stream, content_length

requirements.txt

@@ -5,3 +5,4 @@ httpie
 orjson==3.9.2
 msgpack-python==0.5.6
 pytest==7.4.0
+coverage

tests/test_app.py

@@ -1,13 +1,25 @@
+import io
+from wsgiref.validate import InputWrapper
+
+from unittest.mock import call, MagicMock, mock_open
+
 import falcon
 from falcon import testing
 import msgpack
 import pytest
 
-from app.app import app
+from app.app import create_app
+from app.images import ImageStore
+
+
+@pytest.fixture
+def mock_store():
+    return MagicMock()
 
 
 @pytest.fixture()
-def client():
+def client(mock_store):
+    app = create_app(mock_store)
     return testing.TestClient(app)
 
 
@@ -25,3 +37,46 @@ def test_list_images(client):
 
     assert result_doc == doc
     assert response.status == falcon.HTTP_OK
+
+
+def test_posted_image(client, mock_store):
+    file_name = 'fake-image-name.xyz'
+    mock_store.save.return_value = file_name
+    image_content_type = 'image/xyz'
+
+    response = client.simulate_post(
+        '/images',
+        body=b'some-fake-bytes',
+        headers={'content-type': image_content_type}
+    )
+
+    assert response.status == falcon.HTTP_CREATED
+    assert response.headers['location'] == '/images/{}'.format(file_name)
+    saver_call = mock_store.save.call_args
+
+    assert isinstance(saver_call[0][0], InputWrapper)
+    assert saver_call[0][1] == image_content_type
+
+
+def test_saving_image(monkeypatch):
+    # This still has some mocks, but they are more localized and do not
+    # have to be monkey-patched into standard library modules (always a
+    # risky business).
+    mock_file_open = mock_open()
+
+    fake_uuid = '123e4567-e89b-12d3-a456-426655440000'
+
+    def mock_uuidgen():
+        return fake_uuid
+
+    fake_image_bytes = b'fake-image-bytes'
+    fake_request_stream = io.BytesIO(fake_image_bytes)
+    storage_path = 'fake-storage-path'
+    store = ImageStore(
+        storage_path,
+        uuidgen=mock_uuidgen,
+        fopen=mock_file_open
+    )
+
+    assert store.save(fake_request_stream, 'image/png') == fake_uuid + '.png'
+    assert call().write(fake_image_bytes) in mock_file_open.mock_calls

tests/test_integration.py

@@ -0,0 +1,26 @@
+import os
+import requests
+
+
+def test_posted_image_gets_saved():
+    file_save_prefix = './images/'
+    location_prefix = '/images/'
+    fake_image_bytes = b'fake-image-bytes'
+
+    response = requests.post(
+        'http://localhost:8000/images',
+        data=fake_image_bytes,
+        headers={'content-type': 'image/png'}
+    )
+
+    assert response.status_code == 201
+    location = response.headers['location']
+    assert location.startswith(location_prefix)
+    image_name = location.replace(location_prefix, '')
+
+    file_path = file_save_prefix + image_name
+
+    with open(file_path, 'rb') as image_file:
+        assert image_file.read() == fake_image_bytes
+
+    os.remove(file_path)