From: David Kerkeslager Date: Thu, 28 Nov 2019 23:06:07 +0000 (-0500) Subject: Change the name of the project from phial to fwx, to avoid pypi conflict X-Git-Url: https://code.kerkeslager.com/?p=fwx;a=commitdiff_plain;h=35ce576aef8ec1012a4b579fc8bd7a20d21996db Change the name of the project from phial to fwx, to avoid pypi conflict --- diff --git a/fwx.py b/fwx.py new file mode 100644 index 0000000..411c077 --- /dev/null +++ b/fwx.py @@ -0,0 +1,265 @@ +import collections +import http.cookies +import json +import urllib.parse + +_Request = collections.namedtuple( + 'Request', + ( + 'env', + 'GET', + 'accept', + 'accept_encoding', + 'accept_language', + 'content', + 'content_length', + 'content_type', + 'cookie', + 'method', + 'path', + 'parameters', + 'query', + 'user_agent', + ) +) + +class Request(_Request): + def __new__(cls, env): + errors = [] + + accept = env.get('HTTP_ACCEPT') + accept_encoding = env.get('HTTP_ACCEPT_ENCODING') + accept_language = env.get('HTTP_ACCEPT_LANGUAGE') + content = env.get('CONTENT', '') + content_type = env.get('CONTENT_TYPE') + method = env.get('REQUEST_METHOD') + path = env.get('PATH_INFO') + query = env.get('QUERY_STRING') + user_agent = env.get('HTTP_USER_AGENT') + + content_length = env.get('CONTENT_LENGTH') + + if content_length == '' or content_length is None: + content_length = 0 + else: + try: + content_length = int(content_length) + except ValueError: + errors.append('Unable to parse Content-Length "{}"'.format(content_length)) + content_length = 0 + + try: + cookie = http.cookies.SimpleCookie(env.get('HTTP_COOKIE')) + except: + cookie = http.cookies.SimpleCookie() + + + try: + GET = urllib.parse.parse_qs(query) + except: + GET = {} + errors.append('Unable to parse GET parameters from query string "{}"'.format(query)) + + if method == 'GET': + parameters = GET + + result = super().__new__( + cls, + env=env, + GET=GET, + accept=accept, + accept_encoding=accept_encoding, + accept_language=accept_language, + content = content, + content_length = content_length, + content_type = content_type, + cookie=cookie, + method=method, + parameters=parameters, + path=path, + query=query, + user_agent=user_agent, + ) + + result.subpath = path + return result + +_Response = collections.namedtuple( + 'Response', + ( + 'status', + 'content_type', + 'extra_headers', + 'content', + ), +) + +class Response(_Response): + def __new__(cls, content, **kwargs): + status = kwargs.pop('status', 200) + assert isinstance(status, int) + + content_type = kwargs.pop('content_type') + assert isinstance(content_type, str) + + extra_headers = kwargs.pop('extra_headers', ()) + assert isinstance(extra_headers, tuple) + + assert len(kwargs) == 0 + + return super().__new__( + cls, + status=status, + content_type=content_type, + extra_headers=extra_headers, + content=content, + ) + + @property + def headers(self): + return ( + ('Content-Type', self.content_type), + ) + +class HTMLResponse(Response): + def __new__(cls, content, **kwargs): + assert 'content_type' not in kwargs + + return super().__new__( + cls, + content, + content_type='text/html', + **kwargs, + ) + +class JSONResponse(Response): + def __new__(cls, content_json, **kwargs): + assert 'content_type' not in kwargs + assert 'content' not in kwargs + + self = super().__new__( + cls, + content=json.dumps(content_json), + content_type='application/json', + **kwargs, + ) + self.content_json = content_json + return self + +class TextResponse(Response): + def __new__(cls, content, **kwargs): + assert 'content_type' not in kwargs + + return super().__new__( + cls, + content, + content_type='text/plain', + **kwargs, + ) + +_RedirectResponse = collections.namedtuple( + 'RedirectResponse', + ( + 'location', + 'permanent', + ), +) + +class RedirectResponse(_RedirectResponse): + def __new__(cls, location, **kwargs): + assert isinstance(location, str) + + permanent = kwargs.pop('permanent', True) + assert isinstance(permanent, bool) + assert len(kwargs) == 0 + + return super().__new__( + cls, + location=location, + permanent=permanent, + ) + + @property + def status(self): + return 308 if self.permanent else 307 + + @property + def headers(self): + return (('Location', self.location),) + + @property + def content(self): + return (b'',) + +REQUEST_METHODS = ( + 'GET', + 'HEAD', + 'POST', + 'PUT', + 'PATCH', + 'DELETE', + 'CONNECT', + 'OPTIONS', + 'TRACE', +) + +def default_method_not_allowed_handler(request): + return Response('') + +def default_options_handler(handlers): + def handler(request): + return Response(','.join(handlers.keys())) + return handler + +def route_on_method(**kwargs): + handlers = {} + for method in REQUEST_METHODS: + if method in kwargs: + handlers[method] = kwargs.pop(method) + + method_not_allowed_handler = kwargs.pop( + 'method_not_allowed', + default_method_not_allowed_handler, + ) + + assert len(kwargs) == 0 + + if 'OPTIONS' not in handlers: + handlers['OPTIONS'] = default_options_handler(handlers) + + def handler(request): + return handlers.get( + request.method.upper(), + method_not_allowed_handler, + )(request) + + return handler + +def _get_status(response): + return { + 200: '200 OK', + 307: '307 Temporary Redirect', + 308: '308 Permanent Redirect', + }[response.status] + +def _get_headers(response): + return list(response.headers) + +def _get_content(response): + content = response.content + + if isinstance(content, bytes): + return (content,) + + if isinstance(content, str): + return (content.encode('utf-8'),) + + return content + +def App(handler): + def app(env, start_fn): + response = handler(Request(env)) + + start_fn(_get_status(response), _get_headers(response)) + return _get_content(response) + return app diff --git a/main.py b/main.py index f0bc2bc..d2d758a 100644 --- a/main.py +++ b/main.py @@ -1,11 +1,11 @@ -import phial +import fwx def handler(request): - return phial.TextResponse( + return fwx.TextResponse( content='Hello, world\n', ) -app = phial.App(handler) +app = fwx.App(handler) if __name__ == '__main__': from twisted.internet import reactor diff --git a/phial.py b/phial.py deleted file mode 100644 index 411c077..0000000 --- a/phial.py +++ /dev/null @@ -1,265 +0,0 @@ -import collections -import http.cookies -import json -import urllib.parse - -_Request = collections.namedtuple( - 'Request', - ( - 'env', - 'GET', - 'accept', - 'accept_encoding', - 'accept_language', - 'content', - 'content_length', - 'content_type', - 'cookie', - 'method', - 'path', - 'parameters', - 'query', - 'user_agent', - ) -) - -class Request(_Request): - def __new__(cls, env): - errors = [] - - accept = env.get('HTTP_ACCEPT') - accept_encoding = env.get('HTTP_ACCEPT_ENCODING') - accept_language = env.get('HTTP_ACCEPT_LANGUAGE') - content = env.get('CONTENT', '') - content_type = env.get('CONTENT_TYPE') - method = env.get('REQUEST_METHOD') - path = env.get('PATH_INFO') - query = env.get('QUERY_STRING') - user_agent = env.get('HTTP_USER_AGENT') - - content_length = env.get('CONTENT_LENGTH') - - if content_length == '' or content_length is None: - content_length = 0 - else: - try: - content_length = int(content_length) - except ValueError: - errors.append('Unable to parse Content-Length "{}"'.format(content_length)) - content_length = 0 - - try: - cookie = http.cookies.SimpleCookie(env.get('HTTP_COOKIE')) - except: - cookie = http.cookies.SimpleCookie() - - - try: - GET = urllib.parse.parse_qs(query) - except: - GET = {} - errors.append('Unable to parse GET parameters from query string "{}"'.format(query)) - - if method == 'GET': - parameters = GET - - result = super().__new__( - cls, - env=env, - GET=GET, - accept=accept, - accept_encoding=accept_encoding, - accept_language=accept_language, - content = content, - content_length = content_length, - content_type = content_type, - cookie=cookie, - method=method, - parameters=parameters, - path=path, - query=query, - user_agent=user_agent, - ) - - result.subpath = path - return result - -_Response = collections.namedtuple( - 'Response', - ( - 'status', - 'content_type', - 'extra_headers', - 'content', - ), -) - -class Response(_Response): - def __new__(cls, content, **kwargs): - status = kwargs.pop('status', 200) - assert isinstance(status, int) - - content_type = kwargs.pop('content_type') - assert isinstance(content_type, str) - - extra_headers = kwargs.pop('extra_headers', ()) - assert isinstance(extra_headers, tuple) - - assert len(kwargs) == 0 - - return super().__new__( - cls, - status=status, - content_type=content_type, - extra_headers=extra_headers, - content=content, - ) - - @property - def headers(self): - return ( - ('Content-Type', self.content_type), - ) - -class HTMLResponse(Response): - def __new__(cls, content, **kwargs): - assert 'content_type' not in kwargs - - return super().__new__( - cls, - content, - content_type='text/html', - **kwargs, - ) - -class JSONResponse(Response): - def __new__(cls, content_json, **kwargs): - assert 'content_type' not in kwargs - assert 'content' not in kwargs - - self = super().__new__( - cls, - content=json.dumps(content_json), - content_type='application/json', - **kwargs, - ) - self.content_json = content_json - return self - -class TextResponse(Response): - def __new__(cls, content, **kwargs): - assert 'content_type' not in kwargs - - return super().__new__( - cls, - content, - content_type='text/plain', - **kwargs, - ) - -_RedirectResponse = collections.namedtuple( - 'RedirectResponse', - ( - 'location', - 'permanent', - ), -) - -class RedirectResponse(_RedirectResponse): - def __new__(cls, location, **kwargs): - assert isinstance(location, str) - - permanent = kwargs.pop('permanent', True) - assert isinstance(permanent, bool) - assert len(kwargs) == 0 - - return super().__new__( - cls, - location=location, - permanent=permanent, - ) - - @property - def status(self): - return 308 if self.permanent else 307 - - @property - def headers(self): - return (('Location', self.location),) - - @property - def content(self): - return (b'',) - -REQUEST_METHODS = ( - 'GET', - 'HEAD', - 'POST', - 'PUT', - 'PATCH', - 'DELETE', - 'CONNECT', - 'OPTIONS', - 'TRACE', -) - -def default_method_not_allowed_handler(request): - return Response('') - -def default_options_handler(handlers): - def handler(request): - return Response(','.join(handlers.keys())) - return handler - -def route_on_method(**kwargs): - handlers = {} - for method in REQUEST_METHODS: - if method in kwargs: - handlers[method] = kwargs.pop(method) - - method_not_allowed_handler = kwargs.pop( - 'method_not_allowed', - default_method_not_allowed_handler, - ) - - assert len(kwargs) == 0 - - if 'OPTIONS' not in handlers: - handlers['OPTIONS'] = default_options_handler(handlers) - - def handler(request): - return handlers.get( - request.method.upper(), - method_not_allowed_handler, - )(request) - - return handler - -def _get_status(response): - return { - 200: '200 OK', - 307: '307 Temporary Redirect', - 308: '308 Permanent Redirect', - }[response.status] - -def _get_headers(response): - return list(response.headers) - -def _get_content(response): - content = response.content - - if isinstance(content, bytes): - return (content,) - - if isinstance(content, str): - return (content.encode('utf-8'),) - - return content - -def App(handler): - def app(env, start_fn): - response = handler(Request(env)) - - start_fn(_get_status(response), _get_headers(response)) - return _get_content(response) - return app diff --git a/setup.py b/setup.py index 905fc49..8d1b9e3 100644 --- a/setup.py +++ b/setup.py @@ -4,19 +4,21 @@ with open('README.md', 'r') as fh: long_description = fh.read() setuptools.setup( - name='phial', + name='fwx', version='0.0.1', - scripts=['phial.py'] , + scripts=['fwx.py'] , author='David Kerkeslager', author_email='kerkeslager+pypi@gmail.com', description='A minimalist functional web framework', long_description=long_description, long_description_content_type='text/markdown', - url='https://github.com/kerkeslager/phial', + url='https://github.com/kerkeslager/fwx', packages=setuptools.find_packages(), classifiers=[ 'Programming Language :: Python :: 3', - 'License :: OSI Approved :: MIT License', + 'License :: OSI Approved :: GNU Affero General Public License v3', + 'Development Status :: 2 - Pre-Alpha', + 'Intended Audience :: Developers', 'Operating System :: OS Independent', ], ) diff --git a/test_phial.py b/test_phial.py deleted file mode 100644 index 672be96..0000000 --- a/test_phial.py +++ /dev/null @@ -1,135 +0,0 @@ -import unittest -from unittest import mock - -import phial - -class RequestTests(unittest.TestCase): - def test_GET(self): - request = phial.Request({ - 'REQUEST_METHOD': 'GET', - 'QUERY_STRING': 'foo=bar&baz=qux', - }) - - self.assertEqual(request.GET['foo'], ['bar']) - self.assertEqual(request.GET['baz'], ['qux']) - - def test_parameters(self): - request = phial.Request({ - 'REQUEST_METHOD': 'GET', - 'QUERY_STRING': 'foo=bar&baz=qux', - }) - - self.assertEqual(request.parameters['foo'], ['bar']) - self.assertEqual(request.parameters['baz'], ['qux']) - -class ResponseTests(unittest.TestCase): - def test_content_can_be_positional_argument(self): - response = phial.Response('Hello, world\n', content_type='text/plain') - - self.assertEqual(response.content, 'Hello, world\n') - - def test_content_can_be_keyword_argument(self): - response = phial.Response(content='Hello, world\n', content_type='text/plain') - - self.assertEqual(response.content, 'Hello, world\n') - - def test_status_defaults_to_200(self): - response = phial.Response( - content_type='text/plain', - content='Hello, world\n', - ) - - self.assertEqual(response.status, 200) - - def test_headers(self): - response = phial.Response( - content_type='text/plain', - content='Hello, world\n', - ) - - self.assertEqual( - response.headers, - ( - ('Content-Type', 'text/plain'), - ), - ) - -class HTMLResponseTests(unittest.TestCase): - def test_sets_content_type(self): - response = phial.HTMLResponse('Hello, world') - self.assertEqual(response.content_type, 'text/html') - -class JSONResponseTests(unittest.TestCase): - def test_sets_content_type(self): - response = phial.JSONResponse({ 'foo': 'bar', 'baz': 42 }) - self.assertEqual(response.content_type, 'application/json') - - def test_sets_content(self): - response = phial.JSONResponse({ 'foo': 'bar', 'baz': 42 }) - self.assertEqual(response.content, '{"foo": "bar", "baz": 42}') - - def test_sets_content_json(self): - response = phial.JSONResponse({ 'foo': 'bar', 'baz': 42 }) - self.assertEqual(response.content_json, {"foo": "bar", "baz": 42}) - -class TextResponseTests(unittest.TestCase): - def test_sets_content_type(self): - response = phial.TextResponse('Hello, world\n') - self.assertEqual(response.content_type, 'text/plain') - -class RedirectResponse(unittest.TestCase): - def test_takes_location_as_positional_argument(self): - response = phial.RedirectResponse('/location') - self.assertEqual(response.location, '/location') - - def test_takes_location_as_keyword_argument(self): - response = phial.RedirectResponse(location='/location') - self.assertEqual(response.location, '/location') - - def test_permanent_defaults_to_true(self): - response = phial.RedirectResponse('/location') - self.assertEqual(response.permanent, True) - - def test_status(self): - self.assertEqual( - phial.RedirectResponse('/location', permanent=True).status, - 308, - ) - self.assertEqual( - phial.RedirectResponse('/location', permanent=False).status, - 307, - ) - - def test_headers(self): - self.assertEqual( - phial.RedirectResponse('/location').headers, - (('Location','/location'),), - ) - - def test_content(self): - self.assertEqual( - phial.RedirectResponse('/location').content, - (b'',), - ) - -class _get_status_Tests(unittest.TestCase): - def test_basic(self): - self.assertEqual(phial._get_status(mock.MagicMock(status=200)), '200 OK') - self.assertEqual(phial._get_status(mock.MagicMock(status=307)), '307 Temporary Redirect') - self.assertEqual(phial._get_status(mock.MagicMock(status=308)), '308 Permanent Redirect') - -class _get_content_Tests(unittest.TestCase): - def test_bytes(self): - self.assertEqual( - phial._get_content(mock.MagicMock(content=b'Hello, world\n')), - (b'Hello, world\n',), - ) - - def test_str(self): - self.assertEqual( - phial._get_content(mock.MagicMock(content='Hello, world\n')), - (b'Hello, world\n',), - ) - -if __name__ == '__main__': - unittest.main()