Add 'last_modified' and 'if_modified_since' properties

Author: magv

aiohttp/web_reqrep.py

@@ -4,13 +4,17 @@
 import binascii
 import cgi
 import collections
+import datetime
 import http.cookies
 import io
 import json
+import math
+import time
 import warnings
 
-from urllib.parse import urlsplit, parse_qsl, unquote
+from email.utils import parsedate
 from types import MappingProxyType
+from urllib.parse import urlsplit, parse_qsl, unquote
 
 from . import hdrs
 from .helpers import reify
@@ -65,6 +69,33 @@ def content_length(self, _CONTENT_LENGTH=hdrs.CONTENT_LENGTH):
         else:
             return int(l)
 
+    @property
+    def if_modified_since(self, _IF_MODIFIED_SINCE=hdrs.IF_MODIFIED_SINCE):
+        """The value of If-Modified-Since HTTP header, or None.
+
+        This header is represented as a `datetime` object.
+        """
+        httpdate = self.headers.get(_IF_MODIFIED_SINCE)
+        if httpdate is not None:
+            timetuple = parsedate(httpdate)
+            if timetuple is not None:
+                return datetime.datetime(*timetuple[:6],
+                                         tzinfo=datetime.timezone.utc)
+        return None
+
+    @property
+    def last_modified(self, _LAST_MODIFIED=hdrs.LAST_MODIFIED):
+        """The value of Last-Modified HTTP header, or None.
+
+        This header is represented as a `datetime` object.
+        """
+        httpdate = self.headers.get(_LAST_MODIFIED)
+        if httpdate is not None:
+            timetuple = parsedate(httpdate)
+            if timetuple is not None:
+                return datetime.datetime(*timetuple[:6],
+                                         tzinfo=datetime.timezone.utc)
+        return None
 
 FileField = collections.namedtuple('Field', 'name filename file content_type')
 
@@ -513,6 +544,25 @@ def charset(self, value):
             self._content_dict['charset'] = str(value).lower()
         self._generate_content_type_header()
 
+    @property
+    def last_modified(self):
+        # Just a placeholder for adding setter
+        return super().last_modified
+
+    @last_modified.setter
+    def last_modified(self, value):
+        if value is None:
+            if hdrs.LAST_MODIFIED in self.headers:
+                del self.headers[hdrs.LAST_MODIFIED]
+        elif isinstance(value, (int, float)):
+            self.headers[hdrs.LAST_MODIFIED] = time.strftime(
+                "%a, %d %b %Y %H:%M:%S GMT", time.gmtime(math.ceil(value)))
+        elif isinstance(value, datetime.datetime):
+            self.headers[hdrs.LAST_MODIFIED] = time.strftime(
+                "%a, %d %b %Y %H:%M:%S GMT", value.utctimetuple())
+        elif isinstance(value, str):
+            self.headers[hdrs.LAST_MODIFIED] = value
+
     def _generate_content_type_header(self, CONTENT_TYPE=hdrs.CONTENT_TYPE):
         params = '; '.join("%s=%s" % i for i in self._content_dict.items())
         if params:

aiohttp/web_urldispatcher.py

@@ -11,7 +11,6 @@
 import inspect
 
 from urllib.parse import urlencode
-from wsgiref.handlers import format_date_time
 
 from . import hdrs
 from .abc import AbstractRouter, AbstractMatchInfo
@@ -178,9 +177,9 @@ def handle(self, request):
             raise HTTPNotFound()
 
         st = os.stat(filepath)
-        mtime = format_date_time(st.st_mtime)
 
-        if request.headers.get(hdrs.IF_MODIFIED_SINCE) == mtime:
+        modsince = request.if_modified_since
+        if modsince is not None and st.st_mtime <= modsince.timestamp():
             raise HTTPNotModified()
 
         ct, encoding = mimetypes.guess_type(filepath)
@@ -191,7 +190,7 @@ def handle(self, request):
         resp.content_type = ct
         if encoding:
             resp.headers[hdrs.CONTENT_ENCODING] = encoding
-        resp.headers[hdrs.LAST_MODIFIED] = mtime
+        resp.last_modified = st.st_mtime
 
         file_size = st.st_size
         single_chunk = file_size < self._chunk_size

docs/web_reference.rst

@@ -198,6 +198,15 @@ first positional parameter.
 
       Returns :class:`int` or ``None`` if *Content-Length* is absent.
 
+   .. attribute:: if_modified_since
+
+      Read-only property that returns the date specified in the
+      *If-Modified-Since* header.
+
+      Returns :class:`datetime.datetime` or ``None`` if
+      *If-Modified-Since* header is absent or is not a valid
+      HTTP date.
+
    .. coroutinemethod:: read()
 
       Read request body, returns :class:`bytes` object with body content.
@@ -503,6 +512,15 @@ StreamResponse
 
       The value converted to lower-case on attribute assigning.
 
+   .. attribute:: last_modified
+
+      *Last-Modified* header for outgoing response.
+
+      This property accepts raw :class:`str` values,
+      :class:`datetime.datetime` objects, Unix timestamps specified
+      as an :class:`int` or a :class:`float` object, and the
+      value ``None`` to unset the header.
+
    .. method:: start(request)
 
       :param aiohttp.web.Request request: HTTP request object, that the

tests/test_web_functional.py

@@ -410,14 +410,14 @@ def go(dirname, filename):
             lastmod = 'Mon, 1 Jan 1990 01:01:01 GMT'
             resp = yield from request('GET', url, loop=self.loop,
                                       headers={'If-Modified-Since': lastmod})
-            self.assertIn(resp.status, (200, 304))
+            self.assertEqual(200, resp.status)
             resp.close()
 
         here = os.path.dirname(__file__)
         filename = 'data.unknown_mime_type'
         self.loop.run_until_complete(go(here, filename))
 
-    def test_static_file_not_modified_since(self):
+    def test_static_file_if_modified_since_future_date(self):
 
         @asyncio.coroutine
         def go(dirname, filename):
@@ -429,9 +429,22 @@ def go(dirname, filename):
             lastmod = 'Fri, 31 Dec 9999 23:59:59 GMT'
             resp = yield from request('GET', url, loop=self.loop,
                                       headers={'If-Modified-Since': lastmod})
-            self.assertEqual(200, resp.status)
+            self.assertEqual(304, resp.status)
             resp.close()
 
+        here = os.path.dirname(__file__)
+        filename = 'data.unknown_mime_type'
+        self.loop.run_until_complete(go(here, filename))
+
+    def test_static_file_if_modified_since_invalid_date(self):
+
+        @asyncio.coroutine
+        def go(dirname, filename):
+            app, _, url = yield from self.create_server(
+                'GET', '/static/' + filename
+            )
+            app.router.add_static('/static', dirname)
+
             lastmod = 'not a valid HTTP-date'
             resp = yield from request('GET', url, loop=self.loop,
                                       headers={'If-Modified-Since': lastmod})

tests/test_web_response.py

@@ -1,4 +1,5 @@
 import asyncio
+import datetime
 import unittest
 from unittest import mock
 from aiohttp import hdrs
@@ -103,6 +104,42 @@ def test_charset_without_content_type(self):
         with self.assertRaises(RuntimeError):
             resp.charset = 'koi8-r'
 
+    def test_last_modified_initial(self):
+        resp = StreamResponse()
+        self.assertIsNone(resp.last_modified)
+
+    def test_last_modified_string(self):
+        resp = StreamResponse()
+
+        dt = datetime.datetime(1990, 1, 2, 3, 4, 5, 0, datetime.timezone.utc)
+        resp.last_modified = 'Mon, 2 Jan 1990 03:04:05 GMT'
+        self.assertEqual(resp.last_modified, dt)
+
+    def test_last_modified_timestamp(self):
+        resp = StreamResponse()
+
+        dt = datetime.datetime(1970, 1, 1, 0, 0, 0, 0, datetime.timezone.utc)
+
+        resp.last_modified = 0
+        self.assertEqual(resp.last_modified, dt)
+
+        resp.last_modified = 0.0
+        self.assertEqual(resp.last_modified, dt)
+
+    def test_last_modified_datetime(self):
+        resp = StreamResponse()
+
+        dt = datetime.datetime(2001, 2, 3, 4, 5, 6, 0, datetime.timezone.utc)
+        resp.last_modified = dt
+        self.assertEqual(resp.last_modified, dt)
+
+    def test_last_modified_reset(self):
+        resp = StreamResponse()
+
+        resp.last_modified = 0
+        resp.last_modified = None
+        self.assertEqual(resp.last_modified, None)
+
     @mock.patch('aiohttp.web_reqrep.ResponseImpl')
     def test_start(self, ResponseImpl):
         req = self.make_request('GET', '/')