Compare commits

...

37 Commits

Author SHA1 Message Date
Miguel Grinberg
2efbd67878 Release 2.1.0 2025-02-04 00:31:06 +00:00
Miguel Grinberg
d807011ad0 user logins 2025-02-04 00:04:55 +00:00
Miguel Grinberg
675c978797 Basic and token authentication support 2025-02-03 20:00:36 +00:00
Miguel Grinberg
cd87abba30 Mount unit tests 2025-02-03 11:06:26 +00:00
Miguel Grinberg
fd7931e1ae Added Request.url_prefix, Reques.subapp and local mounts 2025-02-03 00:33:59 +00:00
Maxi
d487a73c1e add js to sse example (#281) 2025-01-22 23:42:51 +00:00
Miguel Grinberg
d864b81b65 revert to default funding file #nolog 2025-01-06 17:49:09 +00:00
Miguel Grinberg
d7459f23b2 Version 2.0.8.dev0 2024-11-10 22:57:45 +00:00
Miguel Grinberg
32f5e415e7 Release 2.0.7 2024-11-10 22:57:31 +00:00
Miguel Grinberg
c46e429106 Accept responses with just a status code (Fixes #263) 2024-11-10 20:09:05 +00:00
Miguel Grinberg
4eac013087 Accept responses with just a status code (Fixes #263) 2024-11-10 00:35:21 +00:00
dependabot[bot]
496a288064 Bump werkzeug from 3.0.3 to 3.0.6 in /examples/benchmark (#260) #nolog
Bumps [werkzeug](https://github.com/pallets/werkzeug) from 3.0.3 to 3.0.6.
- [Release notes](https://github.com/pallets/werkzeug/releases)
- [Changelog](https://github.com/pallets/werkzeug/blob/main/CHANGES.rst)
- [Commits](https://github.com/pallets/werkzeug/compare/3.0.3...3.0.6)

---
updated-dependencies:
- dependency-name: werkzeug
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-10-26 11:37:08 +01:00
dependabot[bot]
bcd876fcae Bump quart from 0.19.4 to 0.19.7 in /examples/benchmark (#259) #nolog
Bumps [quart](https://github.com/pallets/quart) from 0.19.4 to 0.19.7.
- [Release notes](https://github.com/pallets/quart/releases)
- [Changelog](https://github.com/pallets/quart/blob/main/CHANGES.rst)
- [Commits](https://github.com/pallets/quart/compare/0.19.4...0.19.7)

---
updated-dependencies:
- dependency-name: quart
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-10-26 00:57:50 +01:00
Stanislav Garanzha
5e5fc5e93e Fix urls in docs (#253)
* Fix 404 external links in intro.rst

* Fix broken links in extensions.rst

`examples/cors/cors.py` does not exist

`uvicorn.org` - failed to resolve
2024-08-17 18:41:43 +01:00
Miguel Grinberg
8895af3737 add tox to dev dependencies 2024-08-15 20:40:54 +01:00
Miguel Grinberg
0a021462e0 Better documentation for start_server() method (Fixes #252) 2024-08-15 19:10:34 +01:00
Lukas Kremla
482ab6d5ca Fixed gzip automatic content-type assignment and added automatic compression header configuration (#251)
* Fixed gzip automatic content-type assignment and added automatic compression setting

This implements the fix for detecting the proper content-type even when the file has the ".gz" extension. It further makes sure the compression headers are set properly if a "gz." file is detected, but the compression headers weren't explicitly set by the user.

* Added a test for properly auto-determining mime types and setting content encoding header

* Modified the gzip file header assignments and following tests according to the feedback.

---------

Co-authored-by: Lukáš Kremla <lukas.kremla@bonnel.cz>
2024-08-14 23:02:23 +01:00
dependabot[bot]
5fe06f6bd5 Bump certifi from 2023.11.17 to 2024.7.4 in /examples/benchmark (#244)
Bumps [certifi](https://github.com/certifi/python-certifi) from 2023.11.17 to 2024.7.4.
- [Commits](https://github.com/certifi/python-certifi/compare/2023.11.17...2024.07.04)

---
updated-dependencies:
- dependency-name: certifi
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-07-06 11:17:41 +01:00
dependabot[bot]
c170e840ec Bump urllib3 from 2.1.0 to 2.2.2 in /examples/benchmark (#241) #nolog
Bumps [urllib3](https://github.com/urllib3/urllib3) from 2.1.0 to 2.2.2.
- [Release notes](https://github.com/urllib3/urllib3/releases)
- [Changelog](https://github.com/urllib3/urllib3/blob/main/CHANGES.rst)
- [Commits](https://github.com/urllib3/urllib3/compare/2.1.0...2.2.2)

---
updated-dependencies:
- dependency-name: urllib3
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-06-19 00:12:28 +01:00
Miguel Grinberg
3a39b47ea8 Version 2.0.7.dev0 2024-06-18 23:14:36 +01:00
Miguel Grinberg
53287217ae Release 2.0.6 2024-06-18 23:14:14 +01:00
Miguel Grinberg
6ffb8a8fe9 Cookie path support in session and test client 2024-06-18 20:56:18 +01:00
Miguel Grinberg
0151611fc8 Configurable session cookie options (Fixes #242) 2024-06-18 00:09:44 +01:00
dependabot[bot]
4204db61e5 Bump jinja2 from 3.1.3 to 3.1.4 in /examples/benchmark (#230) #nolog
Bumps [jinja2](https://github.com/pallets/jinja) from 3.1.3 to 3.1.4.
- [Release notes](https://github.com/pallets/jinja/releases)
- [Changelog](https://github.com/pallets/jinja/blob/main/CHANGES.rst)
- [Commits](https://github.com/pallets/jinja/compare/3.1.3...3.1.4)

---
updated-dependencies:
- dependency-name: jinja2
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-05-21 11:40:13 +01:00
dependabot[bot]
12438743a8 Bump werkzeug from 3.0.1 to 3.0.3 in /examples/benchmark (#229) #nolog
Bumps [werkzeug](https://github.com/pallets/werkzeug) from 3.0.1 to 3.0.3.
- [Release notes](https://github.com/pallets/werkzeug/releases)
- [Changelog](https://github.com/pallets/werkzeug/blob/main/CHANGES.rst)
- [Commits](https://github.com/pallets/werkzeug/compare/3.0.1...3.0.3)

---
updated-dependencies:
- dependency-name: werkzeug
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-05-21 11:39:51 +01:00
dependabot[bot]
7cbb1edf59 Bump requests from 2.31.0 to 2.32.0 in /examples/benchmark (#232) #nolog
updated-dependencies:
- dependency-name: requests
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-05-21 11:39:16 +01:00
Miguel Grinberg
dac6df7a7a use codecov token for coverage uploads #nolog 2024-04-28 00:31:53 +01:00
dependabot[bot]
5d6e838f3c Bump gunicorn from 21.2.0 to 22.0.0 in /examples/benchmark (#224) #nolog
Bumps [gunicorn](https://github.com/benoitc/gunicorn) from 21.2.0 to 22.0.0.
- [Release notes](https://github.com/benoitc/gunicorn/releases)
- [Commits](https://github.com/benoitc/gunicorn/compare/21.2.0...22.0.0)

---
updated-dependencies:
- dependency-name: gunicorn
  dependency-type: direct:production
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-04-17 23:52:37 +01:00
dependabot[bot]
563bfdc8f5 Bump idna from 3.6 to 3.7 in /examples/benchmark (#223) #nolog
Bumps [idna](https://github.com/kjd/idna) from 3.6 to 3.7.
- [Release notes](https://github.com/kjd/idna/releases)
- [Changelog](https://github.com/kjd/idna/blob/master/HISTORY.rst)
- [Commits](https://github.com/kjd/idna/compare/v3.6...v3.7)

---
updated-dependencies:
- dependency-name: idna
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2024-04-12 15:16:52 +01:00
Miguel Grinberg
679d8e63b8 Fix docs build #nolog 2024-03-24 19:56:43 +00:00
Miguel Grinberg
4cb155ee41 Improved cookie support in the test client 2024-03-24 19:45:22 +00:00
Miguel Grinberg
dea79c5ce2 Make Session class more reusable 2024-03-23 16:29:36 +00:00
Carlo Colombo
6b1fd61917 removed outdated import from documentation (Fixes #216) 2024-03-15 11:03:10 +00:00
Miguel Grinberg
f6876c0d15 Use @wraps on decorated functions 2024-03-14 00:16:35 +00:00
Hamsanger
904d5fcaa2 Add event ID to the SSE implementation (#213) 2024-03-10 23:52:43 +00:00
Miguel Grinberg
a0ea439def Add roadmap details to readme 2024-03-10 17:15:14 +00:00
Miguel Grinberg
a1801d9a53 Version 2.0.6.dev0 2024-03-09 10:48:18 +00:00
36 changed files with 1802 additions and 143 deletions

3
.github/FUNDING.yml vendored
View File

@@ -1,3 +0,0 @@
github: miguelgrinberg
patreon: miguelgrinberg
custom: https://paypal.me/miguelgrinberg

View File

@@ -64,6 +64,7 @@ jobs:
with:
files: ./coverage.xml
fail_ci_if_error: true
token: ${{ secrets.CODECOV_TOKEN }}
benchmark:
name: benchmark
runs-on: ubuntu-latest

4
.gitignore vendored
View File

@@ -25,6 +25,8 @@ wheels/
.installed.cfg
*.egg
MANIFEST
requirements.txt
requirements-dev.txt
# PyInstaller
# Usually these files are written by a python script from a template
@@ -90,6 +92,8 @@ venv/
ENV/
env.bak/
venv.bak/
.direnv
.envrc
# Spyder project settings
.spyderproject

View File

@@ -1,5 +1,32 @@
# Microdot change log
**Release 2.1.0** - 2025-02-04
- User login support ([commit](https://github.com/miguelgrinberg/microdot/commit/d807011ad006e53e70c4594d7eac04d03bb08681))
- Basic and token authentication support ([commit](https://github.com/miguelgrinberg/microdot/commit/675c9787974da926af446974cd96ef224e0ee27f))
- Added `local` argument to the `app.mount()` method, to define sub-application specific before and after request handlers ([commit](https://github.com/miguelgrinberg/microdot/commit/fd7931e1aec173c60f81dad18c1a102ed8f0e081))
- Added `Request.url_prefix`, `Request.subapp` and local mounts ([commit](https://github.com/miguelgrinberg/microdot/commit/fd7931e1aec173c60f81dad18c1a102ed8f0e081))
- Added a front end to the SSE example [#281](https://github.com/miguelgrinberg/microdot/issues/281) ([commit](https://github.com/miguelgrinberg/microdot/commit/d487a73c1ea5b3467e23907618b348ca52e0235c)) (thanks **Maxi**!)
- Additional ``app.mount()`` unit tests ([commit](https://github.com/miguelgrinberg/microdot/commit/cd87abba30206ec6d3928e0aabacb2fccf7baf70))
**Release 2.0.7** - 2024-11-10
- Accept responses with just a status code [#263](https://github.com/miguelgrinberg/microdot/issues/263) ([commit #1](https://github.com/miguelgrinberg/microdot/commit/4eac013087f807cafa244b8a6b7b0ed4c82ff150) [commit #2](https://github.com/miguelgrinberg/microdot/commit/c46e4291061046f1be13f300dd08645b71c16635))
- Fixed compressed file content-type assignment [#251](https://github.com/miguelgrinberg/microdot/issues/251) ([commit](https://github.com/miguelgrinberg/microdot/commit/482ab6d5ca068d71ea6301f45918946161e9fcc1)) (thanks **Lukas Kremla**!)
- Better documentation for start_server[#252](https://github.com/miguelgrinberg/microdot/issues/252) ([commit](https://github.com/miguelgrinberg/microdot/commit/0a021462e0c42c249d587a2d600f5a21a408adfc))
- Fix URLs in documentation [#253](https://github.com/miguelgrinberg/microdot/issues/253) ([commit](https://github.com/miguelgrinberg/microdot/commit/5e5fc5e93e11cbf6e3dc8036494e8732d1815d3e)) (thanks **Stanislav Garanzha**!)
**Release 2.0.6** - 2024-06-18
- Add event ID to the SSE implementation [#213](https://github.com/miguelgrinberg/microdot/issues/213) ([commit](https://github.com/miguelgrinberg/microdot/commit/904d5fcaa2d19d939a719b8e68c4dee3eb470739)) (thanks **Hamsanger**!)
- Configurable session cookie options [#242](https://github.com/miguelgrinberg/microdot/issues/242) ([commit](https://github.com/miguelgrinberg/microdot/commit/0151611fc84fec450820d673f4c4d70c32c990a7))
- Improved cookie support in the test client ([commit](https://github.com/miguelgrinberg/microdot/commit/4cb155ee411dc2d9c9f15714cb32b25ba79b156a))
- Cookie path support in session extension and test client ([commit](https://github.com/miguelgrinberg/microdot/commit/6ffb8a8fe920111c4d8c16e98715a0d5ee2d1da3))
- Refactor `Session` class to make it more reusable ([commit](https://github.com/miguelgrinberg/microdot/commit/dea79c5ce224dec7858ffef45a42bed442fd3a5a))
- Use `@functools.wraps` on decorated functions ([commit](https://github.com/miguelgrinberg/microdot/commit/f6876c0d154adcae96098405fb6a1fdf1ea4ec28))
- Removed outdated import from documentation [#216](https://github.com/miguelgrinberg/microdot/issues/216) ([commit](https://github.com/miguelgrinberg/microdot/commit/6b1fd6191702e7a9ad934fddfcdd0a3cebea7c94)) (thanks **Carlo Colombo**!)
- Add roadmap details to readme ([commit](https://github.com/miguelgrinberg/microdot/commit/a0ea439def238084c4d68309c0992b66ffd28ad6))
**Release 2.0.5** - 2024-03-09
- Correct handling of 0 as an integer argument (regression from #207) [#212](https://github.com/miguelgrinberg/microdot/issues/212) ([commit](https://github.com/miguelgrinberg/microdot/commit/d0a4cf8fa7dfb1da7466157b18d3329a8cf9a5df))

View File

@@ -32,10 +32,25 @@ describes the backwards incompatible changes that were made.
## Resources
- Documentation
- [Stable](https://microdot.readthedocs.io/en/stable/)
- [Latest](https://microdot.readthedocs.io/en/latest/)
- Still using version 1?
- [Code](https://github.com/miguelgrinberg/microdot/tree/v1)
- [Documentation](https://microdot.readthedocs.io/en/v1/)
- [Change Log](https://github.com/miguelgrinberg/microdot/blob/main/CHANGES.md)
- Documentation
- [Latest](https://microdot.readthedocs.io/en/latest/)
- [Stable (v2)](https://microdot.readthedocs.io/en/stable/)
- [Legacy (v1)](https://microdot.readthedocs.io/en/v1/) ([Code](https://github.com/miguelgrinberg/microdot/tree/v1))
## Roadmap
The following features are planned for future releases of Microdot, both for
MicroPython and CPython:
- Support for forms encoded in `multipart/form-data` format
- Authentication support, similar to [Flask-Login](https://github.com/maxcountryman/flask-login) for Flask
- OpenAPI integration, similar to [APIFairy](https://github.com/miguelgrinberg/apifairy) for Flask
In addition to the above, the following extensions are also under consideration,
but only for CPython:
- Database integration through [SQLAlchemy](https://github.com/sqlalchemy/sqlalchemy)
- Socket.IO support through [python-socketio](https://github.com/miguelgrinberg/python-socketio)
Do you have other ideas to propose? Let's [discuss them](https://github.com/miguelgrinberg/microdot/discussions/new?category=ideas)!

View File

@@ -44,6 +44,22 @@ User Sessions
.. automodule:: microdot.session
:members:
Authentication
--------------
.. automodule:: microdot.auth
:inherited-members:
:special-members: __call__
:members:
User Logins
-----------
.. automodule:: microdot.login
:inherited-members:
:special-members: __call__
:members:
Cross-Origin Resource Sharing (CORS)
------------------------------------

View File

@@ -5,8 +5,8 @@ Microdot is a highly extensible web application framework. The extensions
described in this section are maintained as part of the Microdot project in
the same source code repository.
WebSocket Support
~~~~~~~~~~~~~~~~~
WebSocket
~~~~~~~~-
.. list-table::
:align: left
@@ -39,8 +39,8 @@ Example::
message = await ws.receive()
await ws.send(message)
Server-Sent Events Support
~~~~~~~~~~~~~~~~~~~~~~~~~~
Server-Sent Events
~~~~~~~~~~~~~~~~~~
.. list-table::
:align: left
@@ -78,8 +78,8 @@ Example::
the SSE object. For bidirectional communication with the client, use the
WebSocket extension.
Rendering Templates
~~~~~~~~~~~~~~~~~~~
Templates
~~~~~~~~~
Many web applications use HTML templates for rendering content to clients.
Microdot includes extensions to render templates with the
@@ -202,8 +202,8 @@ must be used.
.. note::
The Jinja extension is not compatible with MicroPython.
Maintaining Secure User Sessions
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Secure User Sessions
~~~~~~~~~~~~~~~~~~~~
.. list-table::
:align: left
@@ -270,6 +270,178 @@ The :func:`save() <microdot.session.SessionDict.save>` and
:func:`delete() <microdot.session.SessionDict.delete>` methods are used to update
and destroy the user session respectively.
Authentication
~~~~~~~~~~~~~~
.. list-table::
:align: left
* - Compatibility
- | CPython & MicroPython
* - Required Microdot source files
- | `auth.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot/auth.py>`_
* - Required external dependencies
- | None
* - Examples
- | `basic_auth.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/auth/basic_auth.py>`_
| `token_auth.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/auth/token_auth.py>`_
The authentication extension provides helper classes for two commonly used
authentication patterns, described below.
Basic Authentication
^^^^^^^^^^^^^^^^^^^^
`Basic Authentication <https://en.wikipedia.org/wiki/Basic_access_authentication>`_
is a method of authentication that is part of the HTTP specification. It allows
clients to authenticate to a server using a username and a password. Web
browsers have native support for Basic Authentication and will automatically
prompt the user for a username and a password when a protected resource is
accessed.
To use Basic Authentication, create an instance of the :class:`BasicAuth <microdot.auth.BasicAuth>`
class::
from microdot.auth import BasicAuth
auth = BasicAuth(app)
Next, create an authentication function. The function must accept a request
object and a username and password pair provided by the user. If the
credentials are valid, the function must return an object that represents the
user. If the authentication function cannot validate the user provided
credentials it must return ``None``. Decorate the function with
``@auth.authenticate``::
@auth.authenticate
async def verify_user(request, username, password):
user = await load_user_from_database(username)
if user and user.verify_password(password):
return user
To protect a route with authentication, add the ``auth`` instance as a
decorator::
@app.route('/')
@auth
async def index(request):
return f'Hello, {request.g.current_user}!'
While running an authenticated request, the user object returned by the
authenticaction function is accessible as ``request.g.current_user``.
Token Authentication
^^^^^^^^^^^^^^^^^^^^
To set up token authentication, create an instance of :class:`TokenAuth <microdot.auth.TokenAuth>`::
from microdot.auth import TokenAuth
auth = TokenAuth()
Then add a function that verifies the token and returns the user it belongs to,
or ``None`` if the token is invalid or expired::
@auth.authenticate
async def verify_token(request, token):
return load_user_from_token(token)
As with Basic authentication, the ``auth`` instance is used as a decorator to
protect your routes::
@app.route('/')
@auth
async def index(request):
return f'Hello, {request.g.current_user}!'
User Logins
~~~~~~~~~~~
.. list-table::
:align: left
* - Compatibility
- | CPython & MicroPython
* - Required Microdot source files
- | `login.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot/auth.py>`_
| `session.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot/session.py>`_
* - Required external dependencies
- | CPython: `PyJWT <https://pyjwt.readthedocs.io/>`_
| MicroPython: `jwt.py <https://github.com/micropython/micropython-lib/blob/master/python-ecosys/pyjwt/jwt.py>`_,
`hmac.py <https://github.com/micropython/micropython-lib/blob/master/python-stdlib/hmac/hmac.py>`_
* - Examples
- | `login.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/login/login.py>`_
The login extension provides user login functionality. The logged in state of
the user is stored in the user session cookie, and an optional "remember me"
cookie can also be added to keep the user logged in across browser sessions.
To use this extension, create instances of the
:class:`Session <microdot.session.Session>` and :class:`Login <microdot.login.Login>`
class::
Session(app, secret_key='top-secret!')
login = Login()
The ``Login`` class accept an optional argument with the URL of the login page.
The default for this URL is */login*.
The application must represent users as objects with an ``id`` attribute. A
function decorated with ``@login.user_loader`` is used to load a user object::
@login.user_loader
async def get_user(user_id):
return database.get_user(user_id)
The application must implement the login form. At the point in which the user
credentials have been received and verified, a call to the
:func:`login_user() <microdot.login.Login.login_user>` function must be made to
record the user in the user session::
@app.route('/login', methods=['GET', 'POST'])
async def login(request):
# ...
if user.check_password(password):
return await login.login_user(request, user, remember=remember_me)
return redirect('/login')
The optional ``remember`` argument is used to add a remember me cookie that
will log the user in automatically in future sessions. A value of ``True`` will
keep the log in active for 30 days. Alternatively, an integer number of days
can be passed in this argument.
Any routes that require the user to be logged in must be decorated with
:func:`@login <microdot.login.Login.__call__>`::
@app.route('/')
@login
async def index(request):
# ...
Routes that are of a sensitive nature can be decorated with
:func:`@login.fresh <microdot.login.Login.fresh>`
instead. This decorator requires that the user has logged in during the current
session, and will ask the user to logged in again if the session was
authenticated through a remember me cookie::
@app.get('/fresh')
@login.fresh
async def fresh(request):
# ...
To log out a user, the :func:`logout_user() <microdot.auth.Login.logout_user>`
is used::
@app.post('/logout')
@login
async def logout(request):
await login.logout_user(request)
return redirect('/')
Cross-Origin Resource Sharing (CORS)
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -286,7 +458,7 @@ Cross-Origin Resource Sharing (CORS)
- | None
* - Examples
- | `cors.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/cors/cors.py>`_
- | `app.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/cors/app.py>`_
The CORS extension provides support for `Cross-Origin Resource Sharing
(CORS) <https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS>`_. CORS is a
@@ -305,8 +477,8 @@ Example::
cors = CORS(app, allowed_origins=['https://example.com'],
allow_credentials=True)
Testing with the Test Client
~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Test Client
~~~~~~~~~~~
.. list-table::
:align: left
@@ -342,8 +514,8 @@ Example::
See the documentation for the :class:`TestClient <microdot.test_client.TestClient>`
class for more details.
Deploying on a Production Web Server
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
Production Deployments
~~~~~~~~~~~~~~~~~~~~~~
The ``Microdot`` class creates its own simple web server. This is enough for an
application deployed with MicroPython, but when using CPython it may be useful
@@ -363,7 +535,7 @@ Using an ASGI Web Server
- | `asgi.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot/asgi.py>`_
* - Required external dependencies
- | An ASGI web server, such as `Uvicorn <https://uvicorn.org/>`_.
- | An ASGI web server, such as `Uvicorn <https://www.uvicorn.org/>`_.
* - Examples
- | `hello_asgi.py <https://github.com/miguelgrinberg/microdot/blob/main/examples/hello/hello_asgi.py>`_

View File

@@ -82,8 +82,34 @@ handler functions can be defined as ``async def`` or ``def`` functions, but
``async def`` functions are recommended for performance.
The :func:`run() <microdot.Microdot.run>` method starts the application's web
server on port 5000 by default. This method blocks while it waits for
connections from clients.
server on port 5000 by default, and creates its own asynchronous loop. This
method blocks while it waits for connections from clients.
For some applications it may be necessary to run the web server alongside other
asynchronous tasks, on an already running loop. In that case, instead of
``app.run()`` the web server can be started by invoking the
:func:`start_server() <microdot.Microdot.start_server>` coroutine as shown in
the following example::
import asyncio
from microdot import Microdot
app = Microdot()
@app.route('/')
async def index(request):
return 'Hello, world!'
async def main():
# start the server in a background task
server = asyncio.create_task(app.start_server())
# ... do other asynchronous work here ...
# cleanup before ending the application
await server
asyncio.run(main())
Running with CPython
^^^^^^^^^^^^^^^^^^^^
@@ -92,7 +118,7 @@ Running with CPython
:align: left
* - Required Microdot source files
- | `microdot.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot.py>`_
- | `microdot.py <https://github.com/miguelgrinberg/microdot/blob/main/src/microdot/microdot.py>`_
* - Required external dependencies
- | None
@@ -118,7 +144,7 @@ Running with MicroPython
:align: left
* - Required Microdot source files
- | `microdot.py <https://github.com/miguelgrinberg/microdot/tree/main/src/microdot.py>`_
- | `microdot.py <https://github.com/miguelgrinberg/microdot/blob/main/src/microdot/microdot.py>`_
* - Required external dependencies
- | None
@@ -145,8 +171,9 @@ changed by passing the ``port`` argument to the ``run()`` method.
Web Server Configuration
^^^^^^^^^^^^^^^^^^^^^^^^
The :func:`run() <microdot.Microdot.run>` method supports a few arguments to
configure the web server.
The :func:`run() <microdot.Microdot.run>` and
:func:`start_server() <microdot.Microdot.start_server>` methods support a few
arguments to configure the web server.
- ``port``: The port number to listen on. Pass the desired port number in this
argument to use a port different than the default of 5000. For example::
@@ -418,7 +445,7 @@ Mounting a Sub-Application
^^^^^^^^^^^^^^^^^^^^^^^^^^
Small Microdot applications can be written as a single source file, but this
is not the best option for applications that past a certain size. To make it
is not the best option for applications that pass a certain size. To make it
simpler to write large applications, Microdot supports the concept of
sub-applications that can be "mounted" on a larger application, possibly with
a common URL prefix applied to all of its routes. For developers familiar with
@@ -474,11 +501,25 @@ The resulting application will have the customer endpoints available at
*/customers/* and the order endpoints available at */orders/*.
.. note::
Before-request, after-request and error handlers defined in the
sub-application are also copied over to the main application at mount time.
Once installed in the main application, these handlers will apply to the
whole application and not just the sub-application in which they were
created.
During the handling of a request, the
:attr:`Request.url_prefix <microdot.Microdot.url_prefix>` attribute is
set to the URL prefix under which the sub-application was mounted, or an
empty string if the endpoint did not come from a sub-application or the
sub-application was mounted without a URL prefix. It is possible to issue a
redirect that is relative to the sub-application as follows::
return redirect(request.url_prefix + '/relative-url')
When mounting an application as shown above, before-request, after-request and
error handlers defined in the sub-application are copied over to the main
application at mount time. Once installed in the main application, these
handlers will apply to the whole application and not just the sub-application
in which they were created.
The :func:`mount() <microdot.Microdot.mount>` method has a ``local`` argument
that defaults to ``False``. When this argument is set to ``True``, the
before-request, after-request and error handlers defined in the sub-application
will only apply to the sub-application.
Shutting Down the Server
^^^^^^^^^^^^^^^^^^^^^^^^

1
examples/auth/README.md Normal file
View File

@@ -0,0 +1 @@
This directory contains examples that demonstrate basic and token authentication.

View File

@@ -0,0 +1,31 @@
from microdot import Microdot
from microdot.auth import BasicAuth
from pbkdf2 import generate_password_hash, check_password_hash
# this example provides an implementation of the generate_password_hash and
# check_password_hash functions that can be used in MicroPython. On CPython
# there are many other options for password hashisng so there is no need to use
# this custom solution.
USERS = {
'susan': generate_password_hash('hello'),
'david': generate_password_hash('bye'),
}
app = Microdot()
auth = BasicAuth()
@auth.authenticate
async def check_credentials(request, username, password):
if username in USERS and check_password_hash(USERS[username], password):
return username
@app.route('/')
@auth
async def index(request):
return f'Hello, {request.g.current_user}!'
if __name__ == '__main__':
app.run(debug=True)

47
examples/auth/pbkdf2.py Normal file
View File

@@ -0,0 +1,47 @@
import os
import hashlib
# PBKDF2 secure password hashing algorithm obtained from:
# https://codeandlife.com/2023/01/06/how-to-calculate-pbkdf2-hmac-sha256-with-
# python,-example-code/
def sha256(b):
return hashlib.sha256(b).digest()
def ljust(b, n, f):
return b + f * (n - len(b))
def gethmac(key, content):
okeypad = bytes(v ^ 0x5c for v in ljust(key, 64, b'\0'))
ikeypad = bytes(v ^ 0x36 for v in ljust(key, 64, b'\0'))
return sha256(okeypad + sha256(ikeypad + content))
def pbkdf2(pwd, salt, iterations=1000):
U = salt + b'\x00\x00\x00\x01'
T = bytes(64)
for _ in range(iterations):
U = gethmac(pwd, U)
T = bytes(a ^ b for a, b in zip(U, T))
return T
# The number of iterations may need to be adjusted depending on the hardware.
# Lower numbers make the password hashing algorithm faster but less secure, so
# the largest number that can be tolerated should be used.
def generate_password_hash(password, salt=None, iterations=100000):
salt = salt or os.urandom(16)
dk = pbkdf2(password.encode(), salt, iterations)
return f'pbkdf2-hmac-sha256:{salt.hex()}:{iterations}:{dk.hex()}'
def check_password_hash(password_hash, password):
algorithm, salt, iterations, dk = password_hash.split(':')
iterations = int(iterations)
if algorithm != 'pbkdf2-hmac-sha256':
return False
return pbkdf2(password.encode(), salt=bytes.fromhex(salt),
iterations=iterations) == bytes.fromhex(dk)

View File

@@ -0,0 +1,26 @@
from microdot import Microdot
from microdot.auth import TokenAuth
app = Microdot()
auth = TokenAuth()
TOKENS = {
'susan-token': 'susan',
'david-token': 'david',
}
@auth.authenticate
async def check_token(request, token):
if token in TOKENS:
return TOKENS[token]
@app.route('/')
@auth
async def index(request):
return f'Hello, {request.g.current_user}!'
if __name__ == '__main__':
app.run(debug=True)

View File

@@ -16,7 +16,7 @@ blinker==1.7.0
# quart
build==1.0.3
# via pip-tools
certifi==2023.11.17
certifi==2024.7.4
# via requests
charset-normalizer==3.3.2
# via requests
@@ -32,7 +32,7 @@ flask==3.0.0
# via
# -r requirements.in
# quart
gunicorn==21.2.0
gunicorn==22.0.0
# via -r requirements.in
h11==0.14.0
# via
@@ -49,7 +49,7 @@ hypercorn==0.15.0
# via quart
hyperframe==6.0.1
# via h2
idna==3.6
idna==3.7
# via
# anyio
# requests
@@ -57,7 +57,7 @@ itsdangerous==2.1.2
# via
# flask
# quart
jinja2==3.1.3
jinja2==3.1.4
# via
# flask
# quart
@@ -82,9 +82,9 @@ pydantic-core==2.14.5
# via pydantic
pyproject-hooks==1.0.0
# via build
quart==0.19.4
quart==0.19.7
# via -r requirements.in
requests==2.31.0
requests==2.32.0
# via -r requirements.in
sniffio==1.3.0
# via anyio
@@ -95,11 +95,11 @@ typing-extensions==4.9.0
# fastapi
# pydantic
# pydantic-core
urllib3==2.1.0
urllib3==2.2.2
# via requests
uvicorn==0.24.0.post1
# via -r requirements.in
werkzeug==3.0.1
werkzeug==3.0.6
# via
# flask
# quart

1
examples/login/README.md Normal file
View File

@@ -0,0 +1 @@
This directory contains examples that demonstrate user logins.

123
examples/login/login.py Normal file
View File

@@ -0,0 +1,123 @@
from microdot import Microdot, redirect
from microdot.session import Session
from microdot.login import Login
from pbkdf2 import generate_password_hash, check_password_hash
# this example provides an implementation of the generate_password_hash and
# check_password_hash functions that can be used in MicroPython. On CPython
# there are many other options for password hashisng so there is no need to use
# this custom solution.
class User:
def __init__(self, id, username, password):
self.id = id
self.username = username
self.password_hash = self.create_hash(password)
def create_hash(self, password):
return generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
USERS = {
'user001': User('user001', 'susan', 'hello'),
'user002': User('user002', 'david', 'bye'),
}
app = Microdot()
Session(app, secret_key='top-secret!')
login = Login()
@login.user_loader
async def get_user(user_id):
return USERS.get(user_id)
@app.route('/login', methods=['GET', 'POST'])
async def login_page(request):
if request.method == 'GET':
return '''
<!doctype html>
<html>
<body>
<h1>Please Login</h1>
<form method="POST">
<p>
Username<br>
<input name="username" autofocus>
</p>
<p>
Password:<br>
<input name="password" type="password">
<br>
</p>
<p>
<input name="remember_me" type="checkbox"> Remember me
<br>
</p>
<p>
<button type="submit">Login</button>
</p>
</form>
</body>
</html>
''', {'Content-Type': 'text/html'}
username = request.form['username']
password = request.form['password']
remember_me = bool(request.form.get('remember_me'))
for user in USERS.values():
if user.username == username:
if user.check_password(password):
return await login.login_user(request, user,
remember=remember_me)
return redirect('/login')
@app.route('/')
@login
async def index(request):
return f'''
<!doctype html>
<html>
<body>
<h1>Hello, {request.g.current_user.username}!</h1>
<p>
<a href="/fresh">Click here</a> to access the fresh login page.
</p>
<form method="POST" action="/logout">
<button type="submit">Logout</button>
</form>
</body>
</html>
''', {'Content-Type': 'text/html'}
@app.get('/fresh')
@login.fresh
async def fresh(request):
return f'''
<!doctype html>
<html>
<body>
<h1>Hello, {request.g.current_user.username}!</h1>
<p>This page requires a fresh login session.</p>
<p><a href="/">Go back</a> to the main page.</p>
</body>
</html>
''', {'Content-Type': 'text/html'}
@app.post('/logout')
@login
async def logout(request):
await login.logout_user(request)
return redirect('/')
if __name__ == '__main__':
app.run(debug=True)

47
examples/login/pbkdf2.py Normal file
View File

@@ -0,0 +1,47 @@
import os
import hashlib
# PBKDF2 secure password hashing algorithm obtained from:
# https://codeandlife.com/2023/01/06/how-to-calculate-pbkdf2-hmac-sha256-with-
# python,-example-code/
def sha256(b):
return hashlib.sha256(b).digest()
def ljust(b, n, f):
return b + f * (n - len(b))
def gethmac(key, content):
okeypad = bytes(v ^ 0x5c for v in ljust(key, 64, b'\0'))
ikeypad = bytes(v ^ 0x36 for v in ljust(key, 64, b'\0'))
return sha256(okeypad + sha256(ikeypad + content))
def pbkdf2(pwd, salt, iterations=1000):
U = salt + b'\x00\x00\x00\x01'
T = bytes(64)
for _ in range(iterations):
U = gethmac(pwd, U)
T = bytes(a ^ b for a, b in zip(U, T))
return T
# The number of iterations may need to be adjusted depending on the hardware.
# Lower numbers make the password hashing algorithm faster but less secure, so
# the largest number that can be tolerated should be used.
def generate_password_hash(password, salt=None, iterations=100000):
salt = salt or os.urandom(16)
dk = pbkdf2(password.encode(), salt, iterations)
return f'pbkdf2-hmac-sha256:{salt.hex()}:{iterations}:{dk.hex()}'
def check_password_hash(password_hash, password):
algorithm, salt, iterations, dk = password_hash.split(':')
iterations = int(iterations)
if algorithm != 'pbkdf2-hmac-sha256':
return False
return pbkdf2(password.encode(), salt=bytes.fromhex(salt),
iterations=iterations) == bytes.fromhex(dk)

View File

@@ -1,3 +1,6 @@
# This is a simple example that demonstrates how to use the user session, but
# is not intended as a complete login solution. See the login subdirectory for
# a more complete example.
from microdot import Microdot, Response, redirect
from microdot.session import Session, with_session

View File

@@ -1,16 +1,28 @@
import asyncio
from microdot import Microdot
from microdot import Microdot, send_file
from microdot.sse import with_sse
app = Microdot()
@app.route("/")
async def main(request):
return send_file('index.html')
@app.route('/events')
@with_sse
async def events(request, sse):
for i in range(10):
await asyncio.sleep(1)
await sse.send({'counter': i})
print('Client connected')
try:
i = 0
while True:
await asyncio.sleep(1)
i += 1
await sse.send({'counter': i})
except asyncio.CancelledError:
pass
print('Client disconnected')
app.run(debug=True)
app.run()

30
examples/sse/index.html Normal file
View File

@@ -0,0 +1,30 @@
<!DOCTYPE html>
<html>
<head>
<title>Microdot SSE Example</title>
<meta charset="UTF-8">
</head>
<body>
<h1>Microdot SSE Example</h1>
<div id="log"></div>
<script>
const log = (text, color) => {
document.getElementById('log').innerHTML += `<span style="color: ${color}">${text}</span><br>`;
};
const eventSource = new EventSource('/events');
eventSource.onopen = () => {
log('Connection to server opened.', 'black');
};
eventSource.onmessage = (event) => {
log(`Received message: ${event.data}`, 'blue');
};
eventSource.onerror = (event) => {
log(`EventSource failed: ${event.type}`, 'red');
};
</script>
</body>
</html>

View File

@@ -1,6 +1,6 @@
[project]
name = "microdot"
version = "2.0.5"
version = "2.1.0"
authors = [
{ name = "Miguel Grinberg", email = "miguel.grinberg@gmail.com" },
]
@@ -14,6 +14,8 @@ classifiers = [
"Operating System :: OS Independent",
]
requires-python = ">=3.8"
dependencies = [
]
[project.readme]
file = "README.md"
@@ -24,8 +26,12 @@ Homepage = "https://github.com/miguelgrinberg/microdot"
"Bug Tracker" = "https://github.com/miguelgrinberg/microdot/issues"
[project.optional-dependencies]
dev = [
"tox",
]
docs = [
"sphinx",
"pyjwt",
]
[tool.setuptools]

144
src/microdot/auth.py Normal file
View File

@@ -0,0 +1,144 @@
from microdot import abort
from microdot.microdot import invoke_handler
class BaseAuth:
def __init__(self):
self.auth_callback = None
self.error_callback = None
def __call__(self, f):
"""Decorator to protect a route with authentication.
An instance of this class must be used as a decorator on the routes
that need to be protected. Example::
auth = BasicAuth() # or TokenAuth()
@app.route('/protected')
@auth
def protected(request):
# ...
Routes that are decorated in this way will only be invoked if the
authentication callback returned a valid user object, otherwise the
error callback will be executed.
"""
async def wrapper(request, *args, **kwargs):
auth = self._get_auth(request)
if not auth:
return await invoke_handler(self.error_callback, request)
request.g.current_user = await invoke_handler(
self.auth_callback, request, *auth)
if not request.g.current_user:
return await invoke_handler(self.error_callback, request)
return await invoke_handler(f, request, *args, **kwargs)
return wrapper
class BasicAuth(BaseAuth):
"""Basic Authentication.
:param realm: The realm that is displayed when the user is prompted to
authenticate in the browser.
:param charset: The charset that is used to encode the realm.
:param scheme: The authentication scheme. Defaults to 'Basic'.
:param error_status: The error status code to return when authentication
fails. Defaults to 401.
"""
def __init__(self, realm='Please login', charset='UTF-8', scheme='Basic',
error_status=401):
super().__init__()
self.realm = realm
self.charset = charset
self.scheme = scheme
self.error_status = error_status
self.error_callback = self.authentication_error
def _get_auth(self, request):
auth = request.headers.get('Authorization')
if auth and auth.startswith('Basic '):
import binascii
try:
username, password = binascii.a2b_base64(
auth[6:]).decode().split(':', 1)
except Exception: # pragma: no cover
return None
return username, password
def authentication_error(self, request):
return '', self.error_status, {
'WWW-Authenticate': '{} realm="{}", charset="{}"'.format(
self.scheme, self.realm, self.charset)}
def authenticate(self, f):
"""Decorator to configure the authentication callback.
This decorator must be used with a function that accepts the request
object, a username and a password and returns a user object if the
credentials are valid, or ``None`` if they are not. Example::
@auth.authenticate
async def check_credentials(request, username, password):
user = get_user(username)
if user and user.check_password(password):
return get_user(username)
"""
self.auth_callback = f
class TokenAuth(BaseAuth):
"""Token based authentication.
:param header: The name of the header that will contain the token. Defaults
to 'Authorization'.
:param scheme: The authentication scheme. Defaults to 'Bearer'.
:param error_status: The error status code to return when authentication
fails. Defaults to 401.
"""
def __init__(self, header='Authorization', scheme='Bearer',
error_status=401):
super().__init__()
self.header = header
self.scheme = scheme.lower()
self.error_status = error_status
self.error_callback = self.authentication_error
def _get_auth(self, request):
auth = request.headers.get(self.header)
if auth:
if self.header == 'Authorization':
try:
scheme, token = auth.split(' ', 1)
except Exception:
return None
if scheme.lower() == self.scheme:
return (token.strip(),)
else:
return (auth,)
def authenticate(self, f):
"""Decorator to configure the authentication callback.
This decorator must be used with a function that accepts the request
object, a username and a password and returns a user object if the
credentials are valid, or ``None`` if they are not. Example::
@auth.authenticate
async def check_credentials(request, token):
return get_user(token)
"""
self.auth_callback = f
def errorhandler(self, f):
"""Decorator to configure the error callback.
Microdot calls the error callback to allow the application to generate
a custom error response. The default error response is to call
``abort(401)``.
"""
self.error_callback = f
def authentication_error(self, request):
abort(self.error_status)

8
src/microdot/helpers.py Normal file
View File

@@ -0,0 +1,8 @@
try:
from functools import wraps
except ImportError: # pragma: no cover
# MicroPython does not currently implement functools.wraps
def wraps(wrapped):
def _(wrapper):
return wrapper
return _

163
src/microdot/login.py Normal file
View File

@@ -0,0 +1,163 @@
from time import time
from microdot import redirect
from microdot.microdot import urlencode, invoke_handler
class Login:
"""User login support for Microdot.
:param login_url: the URL to redirect to when a login is required. The
default is '/login'.
"""
def __init__(self, login_url='/login'):
self.login_url = login_url
self.user_loader_callback = None
def user_loader(self, f):
"""Decorator to configure the user callback.
The decorated function receives the user ID as an argument and must
return the corresponding user object, or ``None`` if the user ID is
invalid.
"""
self.user_loader_callback = f
def _get_session(self, request):
return request.app._session.get(request)
def _update_remember_cookie(self, request, days, user_id=None):
remember_payload = request.app._session.encode({
'user_id': user_id,
'days': days,
'exp': time() + days * 24 * 60 * 60
})
@request.after_request
async def _set_remember_cookie(request, response):
response.set_cookie('_remember', remember_payload,
max_age=days * 24 * 60 * 60)
return response
def _get_user_id_from_session(self, request):
session = self._get_session(request)
if session and '_user_id' in session:
return session['_user_id']
if '_remember' in request.cookies:
remember_payload = request.app._session.decode(
request.cookies['_remember'])
user_id = remember_payload.get('user_id')
if user_id: # pragma: no branch
self._update_remember_cookie(
request, remember_payload.get('_days', 30), user_id)
session['_user_id'] = user_id
session['_fresh'] = False
session.save()
return user_id
async def _redirect_to_login(self, request):
return '', 302, {'Location': self.login_url + '?next=' + urlencode(
request.url)}
async def login_user(self, request, user, remember=False,
redirect_url='/'):
"""Log a user in.
:param request: the request object
:param user: the user object
:param remember: if the user's logged in state should be remembered
with a cookie after the session ends. Set to the
number of days the remember cookie should last, or to
``True`` to use a default duration of 30 days.
:param redirect_url: the URL to redirect to after login
This call marks the user as logged in by storing their user ID in the
user session. The application must call this method to log a user in
after their credentials have been validated.
The method returns a redirect response, either to the URL the user
originally intended to visit, or if there is no original URL to the URL
specified by the `redirect_url`.
"""
session = self._get_session(request)
session['_user_id'] = user.id
session['_fresh'] = True
session.save()
if remember:
days = 30 if remember is True else int(remember)
self._update_remember_cookie(request, days, session['_user_id'])
next_url = request.args.get('next', redirect_url)
if not next_url.startswith('/'):
next_url = redirect_url
return redirect(next_url)
async def logout_user(self, request):
"""Log a user out.
:param request: the request object
This call removes information about the user's log in from the user
session. If a remember cookie exists, it is removed as well.
"""
session = self._get_session(request)
session.pop('_user_id', None)
session.pop('_fresh', None)
session.save()
if '_remember' in request.cookies:
self._update_remember_cookie(request, 0)
def __call__(self, f):
"""Decorator to protect a route with authentication.
If the user is not logged in, Microdot will redirect to the login page
first. The decorated route will only run after successful login by the
user. If the user is already logged in, the route will run immediately.
Example::
login = Login()
@app.route('/secret')
@login
async def secret(request):
# only accessible to authenticated users
"""
async def wrapper(request, *args, **kwargs):
user_id = self._get_user_id_from_session(request)
if not user_id:
return await self._redirect_to_login(request)
request.g.current_user = await invoke_handler(
self.user_loader_callback, user_id)
if not request.g.current_user:
return await self._redirect_to_login(request)
return await invoke_handler(f, request, *args, **kwargs)
return wrapper
def fresh(self, f):
"""Decorator to protect a route with "fresh" authentication.
This decorator prevents the route from running when the login session
is not fresh. A fresh session is a session that has been created from
direct user interaction with the login page, while a non-fresh session
occurs when a login is restored from a "remember me" cookie. Example::
login = Login()
@app.route('/secret')
@auth.fresh
async def secret(request):
# only accessible to authenticated users
# users logged in via remember me cookie will need to
# re-authenticate
"""
base_wrapper = self.__call__(f)
async def wrapper(request, *args, **kwargs):
session = self._get_session(request)
if session.get('_fresh'):
return await base_wrapper(request, *args, **kwargs)
return await self._redirect_to_login(request)
return wrapper

View File

@@ -329,7 +329,8 @@ class Request:
pass
def __init__(self, app, client_addr, method, url, http_version, headers,
body=None, stream=None, sock=None):
body=None, stream=None, sock=None, url_prefix='',
subapp=None):
#: The application instance to which this request belongs.
self.app = app
#: The address of the client, as a tuple (host, port).
@@ -338,6 +339,12 @@ class Request:
self.method = method
#: The request URL, including the path and query string.
self.url = url
#: The URL prefix, if the endpoint comes from a mounted
#: sub-application, or else ''.
self.url_prefix = url_prefix
#: The sub-application instance, or `None` if this isn't a mounted
#: endpoint.
self.subapp = subapp
#: The path portion of the URL.
self.path = url
#: The query string portion of the URL.
@@ -598,7 +605,7 @@ class Response:
else: # pragma: no cover
http_cookie += '; Expires=' + time.strftime(
'%a, %d %b %Y %H:%M:%S GMT', expires.timetuple())
if max_age:
if max_age is not None:
http_cookie += '; Max-Age=' + str(max_age)
if secure:
http_cookie += '; Secure'
@@ -616,10 +623,10 @@ class Response:
:param cookie: The cookie's name.
:param kwargs: Any cookie opens and flags supported by
``set_cookie()`` except ``expires``.
``set_cookie()`` except ``expires`` and ``max_age``.
"""
self.set_cookie(cookie, '', expires='Thu, 01 Jan 1970 00:00:01 GMT',
**kwargs)
max_age=0, **kwargs)
def complete(self):
if isinstance(self.body, bytes) and \
@@ -774,7 +781,10 @@ class Response:
first.
"""
if content_type is None:
ext = filename.split('.')[-1]
if compressed and filename.endswith('.gz'):
ext = filename[:-3].split('.')[-1]
else:
ext = filename.split('.')[-1]
if ext in Response.types_map:
content_type = Response.types_map[ext]
else:
@@ -888,6 +898,9 @@ class URLPattern():
except ValueError:
return None, None
def __repr__(self): # pragma: no cover
return 'URLPattern: {}'.format(self.url_pattern)
class HTTPException(Exception):
def __init__(self, status_code, reason=None):
@@ -956,7 +969,7 @@ class Microdot:
def decorated(f):
self.url_map.append(
([m.upper() for m in (methods or ['GET'])],
URLPattern(url_pattern), f))
URLPattern(url_pattern), f, '', None))
return f
return decorated
@@ -1124,24 +1137,33 @@ class Microdot:
return f
return decorated
def mount(self, subapp, url_prefix=''):
def mount(self, subapp, url_prefix='', local=False):
"""Mount a sub-application, optionally under the given URL prefix.
:param subapp: The sub-application to mount.
:param url_prefix: The URL prefix to mount the application under.
:param local: When set to ``True``, the before, after and error request
handlers only apply to endpoints defined in the
sub-application. When ``False``, they apply to the entire
application. The default is ``False``.
"""
for methods, pattern, handler in subapp.url_map:
for methods, pattern, handler, _prefix, _subapp in subapp.url_map:
self.url_map.append(
(methods, URLPattern(url_prefix + pattern.url_pattern),
handler))
for handler in subapp.before_request_handlers:
self.before_request_handlers.append(handler)
for handler in subapp.after_request_handlers:
self.after_request_handlers.append(handler)
for handler in subapp.after_error_request_handlers:
self.after_error_request_handlers.append(handler)
for status_code, handler in subapp.error_handlers.items():
self.error_handlers[status_code] = handler
handler, url_prefix + _prefix, _subapp or subapp))
if not local:
for handler in subapp.before_request_handlers:
self.before_request_handlers.append(handler)
subapp.before_request_handlers = []
for handler in subapp.after_request_handlers:
self.after_request_handlers.append(handler)
subapp.after_request_handlers = []
for handler in subapp.after_error_request_handlers:
self.after_error_request_handlers.append(handler)
subapp.after_error_request_handlers = []
for status_code, handler in subapp.error_handlers.items():
self.error_handlers[status_code] = handler
subapp.error_handlers = {}
@staticmethod
def abort(status_code, reason=None):
@@ -1191,7 +1213,7 @@ class Microdot:
Example::
import asyncio
from microdot_asyncio import Microdot
from microdot import Microdot
app = Microdot()
@@ -1268,7 +1290,7 @@ class Microdot:
Example::
from microdot_asyncio import Microdot
from microdot import Microdot
app = Microdot()
@@ -1299,23 +1321,28 @@ class Microdot:
def find_route(self, req):
method = req.method.upper()
if method == 'OPTIONS' and self.options_handler:
return self.options_handler(req)
return self.options_handler(req), '', None
if method == 'HEAD':
method = 'GET'
f = 404
for route_methods, route_pattern, route_handler in self.url_map:
p = ''
s = None
for route_methods, route_pattern, route_handler, url_prefix, subapp \
in self.url_map:
req.url_args = route_pattern.match(req.path)
if req.url_args is not None:
p = url_prefix
s = subapp
if method in route_methods:
f = route_handler
break
else:
f = 405
return f
return f, p, s
def default_options_handler(self, req):
allow = []
for route_methods, route_pattern, route_handler in self.url_map:
for route_methods, route_pattern, _, _, _ in self.url_map:
if route_pattern.match(req.path) is not None:
allow.extend(route_methods)
if 'GET' in allow:
@@ -1346,38 +1373,76 @@ class Microdot:
method=req.method, path=req.path,
status_code=res.status_code))
def get_request_handlers(self, req, attr, local_first=True):
handlers = getattr(self, attr + '_handlers')
local_handlers = getattr(req.subapp, attr + '_handlers') \
if req and req.subapp else []
return local_handlers + handlers if local_first \
else handlers + local_handlers
async def error_response(self, req, status_code, reason=None):
if req and req.subapp and status_code in req.subapp.error_handlers:
return await invoke_handler(
req.subapp.error_handlers[status_code], req)
elif status_code in self.error_handlers:
return await invoke_handler(self.error_handlers[status_code], req)
return reason or 'N/A', status_code
async def dispatch_request(self, req):
after_request_handled = False
if req:
if req.content_length > req.max_content_length:
if 413 in self.error_handlers:
res = await invoke_handler(self.error_handlers[413], req)
else:
res = 'Payload too large', 413
# the request body is larger than allowed
res = await self.error_response(req, 413, 'Payload too large')
else:
f = self.find_route(req)
# find the route in the app's URL map
f, req.url_prefix, req.subapp = self.find_route(req)
try:
res = None
if callable(f):
for handler in self.before_request_handlers:
# invoke the before request handlers
for handler in self.get_request_handlers(
req, 'before_request', False):
res = await invoke_handler(handler, req)
if res:
break
# invoke the endpoint handler
if res is None:
res = await invoke_handler(
f, req, **req.url_args)
res = await invoke_handler(f, req, **req.url_args)
# process the response
if isinstance(res, int):
# an integer response is taken as a status code
# with an empty body
res = '', res
if isinstance(res, tuple):
# handle a tuple response
if isinstance(res[0], int):
# a tuple that starts with an int has an empty
# body
res = ('', res[0],
res[1] if len(res) > 1 else {})
body = res[0]
if isinstance(res[1], int):
# extract the status code and headers (if
# available)
status_code = res[1]
headers = res[2] if len(res) > 2 else {}
else:
# if the status code is missing, assume 200
status_code = 200
headers = res[1]
res = Response(body, status_code, headers)
elif not isinstance(res, Response):
# any other response types are wrapped in a
# Response object
res = Response(res)
for handler in self.after_request_handlers:
# invoke the after request handlers
for handler in self.get_request_handlers(
req, 'after_request', True):
res = await invoke_handler(
handler, req, res) or res
for handler in req.after_request_handlers:
@@ -1385,50 +1450,62 @@ class Microdot:
handler, req, res) or res
after_request_handled = True
elif isinstance(f, dict):
# the response from an OPTIONS request is a dict with
# headers
res = Response(headers=f)
elif f in self.error_handlers:
res = await invoke_handler(self.error_handlers[f], req)
else:
res = 'Not found', f
# if the route is not found, return a 404 or 405
# response as appropriate
res = await self.error_response(req, f, 'Not found')
except HTTPException as exc:
if exc.status_code in self.error_handlers:
res = self.error_handlers[exc.status_code](req)
else:
res = exc.reason, exc.status_code
# an HTTP exception was raised while handling this request
res = await self.error_response(req, exc.status_code,
exc.reason)
except Exception as exc:
# an unexpected exception was raised while handling this
# request
print_exception(exc)
exc_class = None
# invoke the error handler for the exception class if one
# exists
handler = None
res = None
if exc.__class__ in self.error_handlers:
exc_class = exc.__class__
if req.subapp and exc.__class__ in \
req.subapp.error_handlers:
handler = req.subapp.error_handlers[exc.__class__]
elif exc.__class__ in self.error_handlers:
handler = self.error_handlers[exc.__class__]
else:
# walk up the exception class hierarchy to try to find
# a handler
for c in mro(exc.__class__)[1:]:
if c in self.error_handlers:
exc_class = c
if req.subapp and c in req.subapp.error_handlers:
handler = req.subapp.error_handlers[c]
break
if exc_class:
elif c in self.error_handlers:
handler = self.error_handlers[c]
break
if handler:
try:
res = await invoke_handler(
self.error_handlers[exc_class], req, exc)
res = await invoke_handler(handler, req, exc)
except Exception as exc2: # pragma: no cover
print_exception(exc2)
if res is None:
if 500 in self.error_handlers:
res = await invoke_handler(
self.error_handlers[500], req)
else:
res = 'Internal server error', 500
# if there is still no response, issue a 500 error
res = await self.error_response(
req, 500, 'Internal server error')
else:
if 400 in self.error_handlers:
res = await invoke_handler(self.error_handlers[400], req)
else:
res = 'Bad request', 400
# if the request could not be parsed, issue a 400 error
res = await self.error_response(req, 400, 'Bad request')
if isinstance(res, tuple):
res = Response(*res)
elif not isinstance(res, Response):
res = Response(res)
if not after_request_handled:
for handler in self.after_error_request_handlers:
# if the request did not finish due to an error, invoke the after
# error request handler
for handler in self.get_request_handlers(
req, 'after_error_request', True):
res = await invoke_handler(
handler, req, res) or res
res.is_head = (req and req.method == 'HEAD')

View File

@@ -1,7 +1,6 @@
import jwt
from microdot.microdot import invoke_handler
secret_key = None
from microdot.helpers import wraps
class SessionDict(dict):
@@ -30,14 +29,21 @@ class Session:
"""
secret_key = None
def __init__(self, app=None, secret_key=None):
def __init__(self, app=None, secret_key=None, cookie_options=None):
self.secret_key = secret_key
self.cookie_options = cookie_options or {}
if app is not None:
self.initialize(app)
def initialize(self, app, secret_key=None):
def initialize(self, app, secret_key=None, cookie_options=None):
if secret_key is not None:
self.secret_key = secret_key
if cookie_options is not None:
self.cookie_options = cookie_options
if 'path' not in self.cookie_options:
self.cookie_options['path'] = '/'
if 'http_only' not in self.cookie_options:
self.cookie_options['http_only'] = True
app._session = self
def get(self, request):
@@ -57,13 +63,7 @@ class Session:
if session is None:
request.g._session = SessionDict(request, {})
return request.g._session
try:
session = jwt.decode(session, self.secret_key,
algorithms=['HS256'])
except jwt.exceptions.PyJWTError: # pragma: no cover
request.g._session = SessionDict(request, {})
else:
request.g._session = SessionDict(request, session)
request.g._session = SessionDict(request, self.decode(session))
return request.g._session
def update(self, request, session):
@@ -89,12 +89,12 @@ class Session:
if not self.secret_key:
raise ValueError('The session secret key is not configured')
encoded_session = jwt.encode(session, self.secret_key,
algorithm='HS256')
encoded_session = self.encode(session)
@request.after_request
def _update_session(request, response):
response.set_cookie('session', encoded_session, http_only=True)
response.set_cookie('session', encoded_session,
**self.cookie_options)
return response
def delete(self, request):
@@ -117,10 +117,21 @@ class Session:
"""
@request.after_request
def _delete_session(request, response):
response.set_cookie('session', '', http_only=True,
expires='Thu, 01 Jan 1970 00:00:01 GMT')
response.delete_cookie('session', **self.cookie_options)
return response
def encode(self, payload, secret_key=None):
return jwt.encode(payload, secret_key or self.secret_key,
algorithm='HS256')
def decode(self, session, secret_key=None):
try:
payload = jwt.decode(session, secret_key or self.secret_key,
algorithms=['HS256'])
except jwt.exceptions.PyJWTError: # pragma: no cover
return {}
return payload
def with_session(f):
"""Decorator that passes the user session to the route handler.
@@ -136,13 +147,9 @@ def with_session(f):
Note that the decorator does not save the session. To update the session,
call the :func:`session.save() <microdot.session.SessionDict.save>` method.
"""
@wraps(f)
async def wrapper(request, *args, **kwargs):
return await invoke_handler(
f, request, request.app._session.get(request), *args, **kwargs)
for attr in ['__name__', '__doc__', '__module__', '__qualname__']:
try:
setattr(wrapper, attr, getattr(f, attr))
except AttributeError: # pragma: no cover
pass
return wrapper

View File

@@ -1,5 +1,6 @@
import asyncio
import json
from microdot.helpers import wraps
class SSE:
@@ -12,7 +13,7 @@ class SSE:
self.event = asyncio.Event()
self.queue = []
async def send(self, data, event=None):
async def send(self, data, event=None, event_id=None):
"""Send an event to the client.
:param data: the data to send. It can be given as a string, bytes, dict
@@ -20,6 +21,8 @@ class SSE:
Any other types are converted to string before sending.
:param event: an optional event name, to send along with the data. If
given, it must be a string.
:param event_id: an optional event id, to send along with the data. If
given, it must be a string.
"""
if isinstance(data, (dict, list)):
data = json.dumps(data).encode()
@@ -28,6 +31,8 @@ class SSE:
elif not isinstance(data, bytes):
data = str(data).encode()
data = b'data: ' + data + b'\n\n'
if event_id:
data = b'id: ' + event_id.encode() + b'\n' + data
if event:
data = b'event: ' + event.encode() + b'\n' + data
self.queue.append(data)
@@ -99,6 +104,7 @@ def with_sse(f):
# send a named event
await sse.send('hello', event='greeting')
"""
@wraps(f)
async def sse_handler(request, *args, **kwargs):
return sse_response(request, f, *args, **kwargs)

View File

@@ -77,7 +77,7 @@ class TestClient:
The following example shows how to create a test client for an application
and send a test request::
from microdot_asyncio import Microdot
from microdot import Microdot
app = Microdot()
@@ -112,9 +112,13 @@ class TestClient:
headers['Host'] = 'example.com:1234'
return body, headers
def _process_cookies(self, headers):
def _process_cookies(self, path, headers):
cookies = ''
for name, value in self.cookies.items():
if isinstance(value, tuple):
value, cookie_path = value
if not path.startswith(cookie_path):
continue
if cookies:
cookies += '; '
cookies += name + '=' + value
@@ -123,7 +127,7 @@ class TestClient:
headers['Cookie'] += '; ' + cookies
else:
headers['Cookie'] = cookies
return cookies, headers
return headers
def _render_request(self, method, path, headers, body):
request_bytes = '{method} {path} HTTP/1.0\n'.format(
@@ -139,26 +143,45 @@ class TestClient:
for cookie in cookies:
cookie_name, cookie_value = cookie.split('=', 1)
cookie_options = cookie_value.split(';')
path = '/'
delete = False
for option in cookie_options[1:]:
if option.strip().lower().startswith('expires='):
_, e = option.strip().split('=', 1)
option = option.strip().lower()
if option.startswith(
'max-age='): # pragma: no cover
_, age = option.split('=', 1)
try:
age = int(age)
except ValueError: # pragma: no cover
age = 0
if age <= 0:
delete = True
elif option.startswith('expires='):
_, e = option.split('=', 1)
# this is a very limited parser for cookie expiry
# that only detects a cookie deletion request when
# the date is 1/1/1970
if '1 jan 1970' in e.lower(): # pragma: no branch
delete = True
break
elif option.startswith('path='):
_, path = option.split('=', 1)
if delete:
if cookie_name in self.cookies: # pragma: no branch
del self.cookies[cookie_name]
cookie_path = self.cookies[cookie_name][1] \
if isinstance(self.cookies[cookie_name], tuple) \
else '/'
if path == cookie_path:
del self.cookies[cookie_name]
else:
self.cookies[cookie_name] = cookie_options[0]
if path == '/':
self.cookies[cookie_name] = cookie_options[0]
else:
self.cookies[cookie_name] = (cookie_options[0], path)
async def request(self, method, path, headers=None, body=None, sock=None):
headers = headers or {}
body, headers = self._process_body(body, headers)
cookies, headers = self._process_cookies(headers)
headers = self._process_cookies(path, headers)
request_bytes = self._render_request(method, path, headers, body)
if sock:
reader = sock[0]

View File

@@ -2,6 +2,7 @@ import binascii
import hashlib
from microdot import Request, Response
from microdot.microdot import MUTED_SOCKET_ERRORS, print_exception
from microdot.helpers import wraps
class WebSocketError(Exception):
@@ -192,6 +193,7 @@ async def websocket_upgrade(request):
def websocket_wrapper(f, upgrade_function):
@wraps(f)
async def wrapper(request, *args, **kwargs):
ws = await upgrade_function(request)
try:

View File

@@ -9,3 +9,5 @@ from tests.test_sse import * # noqa: F401, F403
from tests.test_cors import * # noqa: F401, F403
from tests.test_utemplate import * # noqa: F401, F403
from tests.test_session import * # noqa: F401, F403
from tests.test_auth import * # noqa: F401, F403
from tests.test_login import * # noqa: F401, F403

1
tests/files/test.txt.gz Normal file
View File

@@ -0,0 +1 @@
foo

125
tests/test_auth.py Normal file
View File

@@ -0,0 +1,125 @@
import asyncio
import binascii
import unittest
from microdot import Microdot
from microdot.auth import BasicAuth, TokenAuth
from microdot.test_client import TestClient
class TestAuth(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_basic_auth(self):
app = Microdot()
basic_auth = BasicAuth()
@basic_auth.authenticate
def authenticate(request, username, password):
if username == 'foo' and password == 'bar':
return {'username': username}
@app.route('/')
@basic_auth
def index(request):
return request.g.current_user['username']
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={
'Authorization': 'Basic ' + binascii.b2a_base64(
b'foo:bar').decode()}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'foo')
res = self._run(client.get('/', headers={
'Authorization': 'Basic ' + binascii.b2a_base64(
b'foo:baz').decode()}))
self.assertEqual(res.status_code, 401)
def test_token_auth(self):
app = Microdot()
token_auth = TokenAuth()
@token_auth.authenticate
def authenticate(request, token):
if token == 'foo':
return 'user'
@app.route('/')
@token_auth
def index(request):
return request.g.current_user
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={
'Authorization': 'Basic foo'}))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={'Authorization': 'foo'}))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={
'Authorization': 'Bearer foo'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
def test_token_auth_custom_header(self):
app = Microdot()
token_auth = TokenAuth(header='X-Auth-Token')
@token_auth.authenticate
def authenticate(request, token):
if token == 'foo':
return 'user'
@app.route('/')
@token_auth
def index(request):
return request.g.current_user
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={
'Authorization': 'Basic foo'}))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={'Authorization': 'foo'}))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={
'Authorization': 'Bearer foo'}))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={
'X-Token-Auth': 'Bearer foo'}))
self.assertEqual(res.status_code, 401)
res = self._run(client.get('/', headers={'X-Auth-Token': 'foo'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
res = self._run(client.get('/', headers={'x-auth-token': 'foo'}))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user')
@token_auth.errorhandler
def error_handler(request):
return {'status_code': 403}, 403
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 403)
self.assertEqual(res.json, {'status_code': 403})

188
tests/test_login.py Normal file
View File

@@ -0,0 +1,188 @@
import asyncio
import unittest
from microdot import Microdot
from microdot.login import Login
from microdot.session import Session
from microdot.test_client import TestClient
class TestLogin(unittest.TestCase):
@classmethod
def setUpClass(cls):
if hasattr(asyncio, 'set_event_loop'):
asyncio.set_event_loop(asyncio.new_event_loop())
cls.loop = asyncio.get_event_loop()
def _run(self, coro):
return self.loop.run_until_complete(coro)
def test_login(self):
app = Microdot()
Session(app, secret_key='secret')
login = Login()
class User:
def __init__(self, id, name):
self.id = id
self.name = name
@login.user_loader
def load_user(user_id):
return User(user_id, f'user{user_id}')
@app.get('/')
@login
def index(request):
return request.g.current_user.name
@app.post('/login')
async def login_route(request):
return await login.login_user(request, User(123, 'user123'))
@app.post('/logout')
async def logout_route(request):
await login.logout_user(request)
return 'ok'
client = TestClient(app)
res = self._run(client.get('/?foo=bar'))
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/login?next=/%3Ffoo%3Dbar')
res = self._run(client.post('/login?next=/%3Ffoo=bar'))
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/?foo=bar')
self.assertEqual(len(res.headers['Set-Cookie']), 1)
self.assertIn('session', client.cookies)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'user123')
res = self._run(client.post('/logout'))
self.assertEqual(res.status_code, 200)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 302)
def test_login_bad_user_id(self):
class User:
def __init__(self, id, name):
self.id = id
self.name = name
app = Microdot()
Session(app, secret_key='secret')
login = Login()
@login.user_loader
def load_user(user_id):
return None
@app.get('/foo')
@login
async def index(request):
return 'ok'
@app.post('/login')
async def login_route(request):
return await login.login_user(request, User(1, 'user'))
client = TestClient(app)
res = self._run(client.post('/login?next=/'))
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/')
res = self._run(client.get('/foo'))
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/login?next=/foo')
def test_login_bad_redirect(self):
class User:
def __init__(self, id, name):
self.id = id
self.name = name
app = Microdot()
Session(app, secret_key='secret')
login = Login()
@login.user_loader
def load_user(user_id):
return user_id
@app.get('/')
@login
async def index(request):
return 'ok'
@app.post('/login')
async def login_route(request):
return await login.login_user(request, User(1, 'user'))
client = TestClient(app)
res = self._run(client.post('/login?next=http://example.com'))
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/')
def test_login_remember(self):
class User:
def __init__(self, id, name):
self.id = id
self.name = name
app = Microdot()
Session(app, secret_key='secret')
login = Login()
@login.user_loader
def load_user(user_id):
return User(user_id, f'user{user_id}')
@app.get('/')
@login
def index(request):
return {'user': request.g.current_user.id}
@app.post('/login')
async def login_route(request):
return await login.login_user(request, User(1, 'user1'),
remember=True)
@app.post('/logout')
async def logout(request):
await login.logout_user(request)
return 'ok'
@app.get('/fresh')
@login.fresh
async def fresh(request):
return f'fresh {request.g.current_user.id}'
client = TestClient(app)
res = self._run(client.post('/login?next=/%3Ffoo=bar'))
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/?foo=bar')
self.assertEqual(len(res.headers['Set-Cookie']), 2)
self.assertIn('session', client.cookies)
self.assertIn('_remember', client.cookies)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, '{"user": 1}')
res = self._run(client.get('/fresh'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'fresh 1')
del client.cookies['session']
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
res = self._run(client.get('/fresh'))
self.assertEqual(res.status_code, 302)
self.assertEqual(res.headers['Location'], '/login?next=/fresh')
res = self._run(client.post('/logout'))
self.assertEqual(res.status_code, 200)
self.assertFalse('_remember' in client.cookies)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 302)

View File

@@ -203,6 +203,7 @@ class TestMicrodot(unittest.TestCase):
req.cookies['one'] + req.cookies['two'] + req.cookies['three'])
res.set_cookie('four', '4')
res.delete_cookie('two', path='/')
res.delete_cookie('one', path='/bad')
return res
client = TestClient(app, cookies={'one': '1', 'two': '2'})
@@ -273,6 +274,14 @@ class TestMicrodot(unittest.TestCase):
return '<p>four</p>', 202, \
{'Content-Type': 'text/html; charset=UTF-8'}
@app.route('/status')
def five(req):
return 202
@app.route('/status-headers')
def six(req):
return 202, {'Content-Type': 'text/html; charset=UTF-8'}
client = TestClient(app)
res = self._run(client.get('/body'))
@@ -298,6 +307,18 @@ class TestMicrodot(unittest.TestCase):
'text/html; charset=UTF-8')
self.assertEqual(res.text, '<p>four</p>')
res = self._run(client.get('/status'))
self.assertEqual(res.text, '')
self.assertEqual(res.status_code, 202)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
res = self._run(client.get('/status-headers'))
self.assertEqual(res.text, '')
self.assertEqual(res.status_code, 202)
self.assertEqual(res.headers['Content-Type'],
'text/html; charset=UTF-8')
def test_before_after_request(self):
app = Microdot()
@@ -773,7 +794,7 @@ class TestMicrodot(unittest.TestCase):
@subapp.route('/app')
def index(req):
return req.g.before + ':foo'
return req.g.before + ':' + req.url_prefix
app = Microdot()
app.mount(subapp, url_prefix='/sub')
@@ -790,4 +811,203 @@ class TestMicrodot(unittest.TestCase):
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'before:foo:after')
self.assertEqual(res.text, 'before:/sub:after')
def test_mount_local(self):
subapp1 = Microdot()
subapp2 = Microdot()
@subapp1.before_request
def before1(req):
req.g.before += ':before1'
@subapp1.after_error_request
def after_error1(req, res):
res.body += b':errorafter'
@subapp1.errorhandler(ValueError)
def value_error(req, exc):
return str(exc), 400
@subapp1.route('/')
def index1(req):
raise ZeroDivisionError()
@subapp1.route('/foo')
def foo(req):
return req.g.before + ':foo:' + req.url_prefix
@subapp1.route('/err')
def err(req):
raise ValueError('err')
@subapp1.route('/err2')
def err2(req):
class MyErr(ValueError):
pass
raise MyErr('err')
@subapp2.before_request
def before2(req):
req.g.before += ':before2'
@subapp2.after_request
def after2(req, res):
res.body += b':after'
@subapp2.errorhandler(405)
def method_not_found2(req):
return '405', 405
@subapp2.route('/bar')
def bar(req):
return req.g.before + ':bar:' + req.url_prefix
@subapp2.route('/baz')
def baz(req):
abort(405)
app = Microdot()
@app.before_request
def before(req):
req.g.before = 'before-app'
@app.after_request
def after(req, res):
res.body += b':after-app'
app.mount(subapp1, local=True)
app.mount(subapp2, url_prefix='/sub', local=True)
client = TestClient(app)
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 500)
self.assertEqual(res.text, 'Internal server error:errorafter')
res = self._run(client.get('/foo'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text, 'before-app:before1:foo::after-app')
res = self._run(client.get('/err'))
self.assertEqual(res.status_code, 400)
self.assertEqual(res.text, 'err:errorafter')
res = self._run(client.get('/err2'))
self.assertEqual(res.status_code, 400)
self.assertEqual(res.text, 'err:errorafter')
res = self._run(client.get('/sub/bar'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'text/plain; charset=UTF-8')
self.assertEqual(res.text,
'before-app:before2:bar:/sub:after:after-app')
res = self._run(client.post('/sub/bar'))
self.assertEqual(res.status_code, 405)
self.assertEqual(res.text, '405')
res = self._run(client.get('/sub/baz'))
self.assertEqual(res.status_code, 405)
self.assertEqual(res.text, '405')
def test_many_mounts(self):
subsubapp = Microdot()
@subsubapp.before_request
def subsubapp_before(req):
req.g.before = 'subsubapp'
@subsubapp.route('/')
def subsubapp_index(req):
return f'{req.g.before}:{req.subapp == subsubapp}:{req.url_prefix}'
subapp = Microdot()
@subapp.before_request
def subapp_before(req):
req.g.before = 'subapp'
@subapp.route('/')
def subapp_index(req):
return f'{req.g.before}:{req.subapp == subapp}:{req.url_prefix}'
app = Microdot()
@app.before_request
def app_before(req):
req.g.before = 'app'
@app.route('/')
def app_index(req):
return f'{req.g.before}:{req.subapp is None}:{req.url_prefix}'
subapp.mount(subsubapp, url_prefix='/subsub')
app.mount(subapp, url_prefix='/sub')
client = TestClient(app)
res = self._run(client.get('/sub/subsub/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'subsubapp:True:/sub/subsub')
res = self._run(client.get('/sub/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'subsubapp:True:/sub')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'subsubapp:True:')
def test_many_local_mounts(self):
subsubapp = Microdot()
@subsubapp.before_request
def subsubapp_before(req):
req.g.before = 'subsubapp'
@subsubapp.route('/')
def subsubapp_index(req):
return f'{req.g.before}:{req.subapp == subsubapp}:{req.url_prefix}'
subapp = Microdot()
@subapp.before_request
def subapp_before(req):
req.g.before = 'subapp'
@subapp.route('/')
def subapp_index(req):
return f'{req.g.before}:{req.subapp == subapp}:{req.url_prefix}'
app = Microdot()
@app.before_request
def app_before(req):
req.g.before = 'app'
@app.route('/')
def app_index(req):
return f'{req.g.before}:{req.subapp is None}:{req.url_prefix}'
subapp.mount(subsubapp, url_prefix='/subsub', local=True)
app.mount(subapp, url_prefix='/sub', local=True)
client = TestClient(app)
res = self._run(client.get('/sub/subsub/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'subsubapp:True:/sub/subsub')
res = self._run(client.get('/sub/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'subapp:True:/sub')
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
self.assertEqual(res.text, 'app:True:')

View File

@@ -136,10 +136,10 @@ class TestResponse(unittest.TestCase):
self.assertTrue(fd.response.endswith(b'\r\n\r\nfoobar'))
def test_create_from_other(self):
res = Response(123)
res = Response(23.7)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers, {})
self.assertEqual(res.body, 123)
self.assertEqual(res.body, 23.7)
def test_create_with_status_code(self):
res = Response('not found', 404)
@@ -193,6 +193,7 @@ class TestResponse(unittest.TestCase):
expires='Tue, 05 Nov 2019 02:23:54 GMT', max_age=123,
secure=True, http_only=True)
res.delete_cookie('foo8', http_only=True)
res.delete_cookie('foo9', path='/s')
self.assertEqual(res.headers, {'Set-Cookie': [
'foo1=bar1',
'foo2=bar2; Path=/; Partitioned',
@@ -203,7 +204,10 @@ class TestResponse(unittest.TestCase):
'foo7=bar7; Path=/foo; Domain=example.com:1234; '
'Expires=Tue, 05 Nov 2019 02:23:54 GMT; Max-Age=123; Secure; '
'HttpOnly',
'foo8=; Expires=Thu, 01 Jan 1970 00:00:01 GMT; HttpOnly',
('foo8=; Expires=Thu, 01 Jan 1970 00:00:01 GMT; Max-Age=0; '
'HttpOnly'),
('foo9=; Path=/s; Expires=Thu, 01 Jan 1970 00:00:01 GMT; '
'Max-Age=0'),
]})
def test_redirect(self):
@@ -276,6 +280,17 @@ class TestResponse(unittest.TestCase):
'application/octet-stream')
self.assertEqual(res.headers['Content-Encoding'], 'gzip')
def test_send_file_gzip_handling(self):
res = Response.send_file('tests/files/test.txt.gz')
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'],
'application/octet-stream')
res = Response.send_file('tests/files/test.txt.gz', compressed=True)
self.assertEqual(res.status_code, 200)
self.assertEqual(res.headers['Content-Type'], 'text/plain')
self.assertEqual(res.headers['Content-Encoding'], 'gzip')
def test_default_content_type(self):
original_content_type = Response.default_content_type
res = Response('foo')

View File

@@ -37,7 +37,7 @@ class TestSession(unittest.TestCase):
@app.post('/set')
@with_session
async def save_session(req, session):
def save_session(req, session):
session['name'] = 'joe'
session.save()
return 'OK'
@@ -82,3 +82,77 @@ class TestSession(unittest.TestCase):
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
def test_session_default_path(self):
app = Microdot()
Session(app, secret_key='some-other-secret')
client = TestClient(app)
@app.get('/')
@with_session
def index(req, session):
session['foo'] = 'bar'
session.save()
return ''
@app.get('/child')
@with_session
def child(req, session):
return str(session.get('foo'))
@app.get('/delete')
@with_session
def delete(req, session):
session.delete()
return ''
res = self._run(client.get('/'))
self.assertEqual(res.status_code, 200)
res = self._run(client.get('/child'))
self.assertEqual(res.text, 'bar')
res = self._run(client.get('/delete'))
res = self._run(client.get('/child'))
self.assertEqual(res.text, 'None')
def test_session_custom_path(self):
app = Microdot()
session_ext = Session()
session_ext.initialize(app, secret_key='some-other-secret',
cookie_options={'path': '/child',
'http_only': False})
client = TestClient(app)
@app.get('/')
@with_session
def index(req, session):
return str(session.get('foo'))
@app.get('/child')
@with_session
def child(req, session):
session['foo'] = 'bar'
session.save()
return ''
@app.get('/child/foo')
@with_session
def foo(req, session):
return str(session.get('foo'))
@app.get('/child/delete')
@with_session
def delete(req, session):
session.delete()
return ''
res = self._run(client.get('/child'))
self.assertEqual(res.status_code, 200)
res = self._run(client.get('/'))
self.assertEqual(res.text, 'None')
res = self._run(client.get('/child/foo'))
self.assertEqual(res.text, 'bar')
res = self._run(client.get('/child/delete'))
res = self._run(client.get('/'))
self.assertEqual(res.text, 'None')
res = self._run(client.get('/child/foo'))
self.assertEqual(res.text, 'None')

View File

@@ -23,6 +23,8 @@ class TestWebSocket(unittest.TestCase):
async def handle_sse(request, sse):
await sse.send('foo')
await sse.send('bar', event='test')
await sse.send('bar', event='test', event_id='id42')
await sse.send('bar', event_id='id42')
await sse.send({'foo': 'bar'})
await sse.send([42, 'foo', 'bar'])
await sse.send(ValueError('foo'))
@@ -34,6 +36,8 @@ class TestWebSocket(unittest.TestCase):
self.assertEqual(response.headers['Content-Type'], 'text/event-stream')
self.assertEqual(response.text, ('data: foo\n\n'
'event: test\ndata: bar\n\n'
'event: test\nid: id42\ndata: bar\n\n'
'id: id42\ndata: bar\n\n'
'data: {"foo": "bar"}\n\n'
'data: [42, "foo", "bar"]\n\n'
'data: foo\n\n'