Compare commits

..

123 Commits

Author SHA1 Message Date
41ba577afa
[release] halfapi 0.6.31 2025-05-17 01:32:10 +02:00
f9bef2b124
[deps] bump versions for pyjwt and schema 2025-05-17 01:27:04 +02:00
ee3bd4413c
[deps] starlette v0.46.2 2025-05-17 01:13:55 +02:00
50a8431ca3
[deps] bump versions in Pipfile.lock, remove mix_stderr with "click" new version, needs orjson>=3.10 2025-05-17 00:59:27 +02:00
c808ba21ab
[fix] oubli import 2024-05-28 03:40:29 +02:00
9f908d3cee
[release] halfapi 0.6.30 2024-05-28 02:55:17 +02:00
7b6d9e994a
[deps] starlette v0.37 2024-05-28 02:54:50 +02:00
8f6330bca7
[deps] starlette 0.35 2024-05-28 02:53:58 +02:00
8506aa5322
[middleware] fix DomainMiddleware constructor for 0.35 2024-05-28 02:41:41 +02:00
b683e80959
[halfapi] remove on_startup argument according to starlette 0.26 2024-05-28 02:41:19 +02:00
b412d249a1
[deps] bump dependencies versions 2024-05-28 02:39:08 +02:00
Maxime Alves LIRMM
f4f9a0fc66 [dockerfile] alpine 3.19 - utilisation d'un venv 2023-12-13 11:37:29 +01:00
Maxime Alves LIRMM
c855cce013 [release] 0.6.29 2023-12-13 11:09:42 +01:00
Maxime Alves LIRMM
e083c4386e [release] 0.6.28 2023-11-24 14:31:10 +01:00
Maxime Alves LIRMM
476ae29792 [rc] 0.6.28rc5 2023-09-01 04:59:17 +09:00
Maxime Alves LIRMM
673097adeb [half_domain] fix ACL route when checked ACL is a decorator 2023-09-01 04:58:12 +09:00
Maxime Alves LIRMM
1cc1bbd5ef [rc] 0.6.28rc4 (remove rc3) 2023-08-21 08:47:47 +02:00
Maxime Alves LIRMM
135d6e86e4 [conf] fix export SECRET variable 2023-08-21 08:47:19 +02:00
Maxime Alves LIRMM
0fcf433ec6 [ci] use alpine base image directly 2023-08-21 00:08:29 +02:00
Maxime Alves LIRMM
45cf32de2b [ci] utilisation de l'image 3.17 (pour fix le bug exceptiongroup) 2023-08-20 23:53:18 +02:00
Maxime Alves LIRMM
1b713c3816 [ci] fix image pour les tests <3.11 2023-08-20 23:42:10 +02:00
Maxime Alves LIRMM
59889e1e31 [ci] pipenv --skip-lock (has been remove) 2023-08-20 23:34:34 +02:00
Maxime Alves LIRMM
28a1a69435 [rc] 0.6.28rc3 - fix bugs and general configuration management cleanup (see changelog) 2023-08-20 23:32:50 +02:00
65ecf9817c [ci/Dockerfile] use of alpine OS instead of debian, bump python to version 3.11 2023-08-09 14:46:52 +02:00
3b7d3bda5c [rc] 0.6.28rc2 2023-08-09 14:24:32 +02:00
e19f27f306 [deps] starlette 0.31 2023-08-09 14:23:09 +02:00
e9c84c9f7c [rc] 0.6.28-rc1 2023-08-08 09:09:16 +02:00
b1595beb14 [tests] write a schema component for dummy_domain for example 2023-08-02 13:29:57 +02:00
60ff99d0fb [domain] you can specify Schema components under the "schema_components" key of the domain dictionary 2023-08-02 12:55:04 +02:00
9657f0f9ec [ci-cd] add python 3.11 to testing releases 2023-08-02 10:57:33 +02:00
f646b4d663 [ci-cd] coverage report 2023-08-02 10:49:54 +02:00
0817882558 [dev-deps] ajout de la dépendance a coverage 2023-08-02 10:43:50 +02:00
896ce58731 [pytest] do not avoid to run doctest modules in halfapi sources! 2023-08-02 10:39:06 +02:00
87856cfb42 [cli-conf] halfapi domain : the file provided as argument is a toml file of the format of .halfapi/config, + better config handling 2023-08-02 10:38:36 +02:00
4856f80b99 [rc] 0.6.28-rc0-1 2023-08-02 06:32:26 +02:00
eac602f0a5 [rc] 0.6.28rc0 2023-08-01 20:50:58 +02:00
14e051bd91 [doc-schema] the "/" route on a domain now returns the OpenAPI-validated Schema (not a list of schemas), the "dummy_domain" test now validates OpenAPI specs 2023-08-01 20:31:17 +02:00
20563081f5 [doc-schema] In module-based routers, if there is a path parameter, you can specify an OpenAPI documentation for it, or a default will be used 2023-08-01 20:24:24 +02:00
7949b3206c [dev-deps] openapi-schema-validator, openapi-spec-validator 2023-08-01 19:41:49 +02:00
c4583b7187 [doc] add docstrings for halfapi routes 2023-08-01 17:43:59 +02:00
2413436104 [acl] The public acls check routes use the "HEAD" method, deprecated "GET" 2023-08-01 17:32:25 +02:00
Maxime Alves LIRMM
54cc6c17c9 [release] 0.6.27 2023-06-07 11:46:53 +02:00
Maxime Alves LIRMM
ff3a39c740 [rc] 0.6.27rc0 2023-06-01 15:39:57 +02:00
Maxime Alves LIRMM@home
8d254bafa0 [feature] changes in the ACLs result availability 2023-06-01 15:39:44 +02:00
Maxime Alves LIRMM
0a385661b9 [rc] 0.6.26rc0 2023-02-23 11:44:57 +01:00
Maxime Alves LIRMM
e065fe04e4 [tests] test with multiple optional parameteres 2023-02-21 19:30:59 +01:00
Maxime Alves LIRMM
b7c5704c95 [tests] fix tests so the data is sent in json 2023-02-21 19:23:23 +01:00
Maxime Alves LIRMM
dd83a337e9 [lib.domain] route_decorator : Adds the "base_url", "cookies" and "url" to the "halfapi" argument of route definitions 2023-02-21 19:22:57 +01:00
Maxime Alves LIRMM
f6d08e8309 [release] 0.6.25 2023-02-20 12:18:38 +01:00
Maxime Alves LIRMM
262de901a8 [jwtMiddleware] on expired signature error, returns Nobody and do not raise an exception 2023-02-13 11:14:24 +01:00
Maxime Alves LIRMM
e5c25ede1f [rc] 0.6.25-rc0 2023-02-08 12:54:35 +01:00
Maxime Alves LIRMM
b4c37ea999 [release] 0.6.24 2023-02-03 14:22:04 +01:00
Maxime Alves LIRMM
5a7e51ae94 [jwtMiddleware] clean "is_fake_user_id" code 2023-02-03 14:20:01 +01:00
Maxime Alves LIRMM
69129fd7af [doc][release] 0.7.0 2023-02-03 12:43:30 +01:00
Maxime Alves LIRMM
a3fc6dc830 [authMiddleware] UN-Breaking uses either the cookie or the header names "Authorization" 2023-02-03 12:43:16 +01:00
Maxime Alves LIRMM@home
064127dc16 [fix] tests for running by implementing the dry-run option to the domain command 2023-02-03 00:02:56 +01:00
Maxime Alves LIRMM@home
c2eb95533c [fix] fix check_domain test c9ecc1f8d2 2023-02-02 22:33:35 +01:00
Maxime Alves LIRMM
bbb027cd0d [authMiddleware][BREAKING] read token from "JWTToken" cookie 2023-02-02 19:56:59 +01:00
Maxime Alves LIRMM
c9ecc1f8d2 [cli] allow to run halfapi on a specified domain using the "halfapi domain --run DOMAIN" command 2023-02-02 19:56:11 +01:00
Maxime Alves LIRMM
d1a8351997 [halfapi] if there is only one domain it is automatically enabled 2023-02-02 19:55:37 +01:00
Maxime Alves LIRMM
dc29abea84 [release] 0.6.23 2023-01-14 12:03:40 +01:00
Maxime Alves LIRMM
4c966e7151 [deps] toml bumps version 0.10 2023-01-14 11:43:38 +01:00
Maxime Alves LIRMM
7deb353b4b [deps] PyYAML bumps version 6.0 2023-01-14 11:39:40 +01:00
Maxime Alves LIRMM
3986083725 [deps] PyJWT bumps version 2.6.0 2023-01-14 11:37:08 +01:00
Maxime Alves LIRMM
301b0eeab5 [deps] click bump version 8 2023-01-14 11:29:14 +01:00
Maxime Alves LIRMM
cc0566542b [deps] orjson bump version 3.8.5 2023-01-14 11:23:25 +01:00
Maxime Alves LIRMM
4f4dac0ff2 [deps] Migration from starlette v0.18 to v0.23
Breaking : migrate your tests that use the TestDomain.client method
following the instructions here https://github.com/Kludex/bump-testclient

Squashed commit of the following:

commit 0417f27b3f
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 11:08:44 2023 +0100

    [deps] starlette 0.23

commit 552f00a65b
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:59:42 2023 +0100

    [deps] starlette 0.22

commit aefe448717
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:55:45 2023 +0100

    [tests][fix] compares the json interpreted value instead of the string

commit 01333a200c
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:55:20 2023 +0100

    [testing] changes from requests to httpx for Starlette TestClient (breaks)

commit f3784fab7f
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:54:10 2023 +0100

    [deps][breaking] starlette 0.21

commit 717d3f8bd6
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:26:31 2023 +0100

    [responses] use a wrapper function for exception handling (fix starlette 0.20)

commit d0876e45da
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:25:21 2023 +0100

    [deps][breaking] starlette 0.20

commit 6504191c53
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:12:51 2023 +0100

    [deps] starlette 0.19

commit 7b639a8dc2
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:11:14 2023 +0100

    [deps] starlette 0.18

commit 20bd9077a4
Author: Maxime Alves LIRMM <maxime.alves@lirmm.fr>
Date:   Sat Jan 14 10:07:48 2023 +0100

    pipenv update
2023-01-14 11:10:13 +01:00
Maxime Alves LIRMM
8b8caa2e15 [release] 0.6.22 2022-12-02 18:11:26 +01:00
Maxime Alves LIRMM
5a70c00541 [rc] 0.6.22-rc6 2022-10-13 12:21:26 +02:00
Maxime Alves LIRMM
7723acb812 [breaking] domain description is defined in a dict, not in "__name__", "__id__", "__routers__" ... 2022-10-13 12:18:13 +02:00
Maxime Alves LIRMM
fd682ba0e0 [fix] wrong "data" argument value when there is a default one 2022-09-13 19:15:45 +02:00
Maxime Alves LIRMM@home
e9c57049dd [lib/domain] log.error unhandled exceptions 2022-09-06 21:51:48 +02:00
Maxime Alves LIRMM
e47ffcddb9 [deps] update pyJWT>2.4.0 2022-09-06 19:58:41 +02:00
Maxime Alves LIRMM
f0c898ba20 (0.6.22rc3) [halfapi] Log HTTPException with statuscode 500 as critical 2022-09-05 10:44:04 +02:00
Maxime Alves LIRMM@home
b7e678e00f [rc] 0.6.22-rc2 2022-09-05 10:20:08 +02:00
Maxime Alves LIRMM@home
d0ec030ce9 [fix] important fix for argument handling 2022-09-05 10:18:57 +02:00
Maxime Alves LIRMM@home
039bc2c8fe [tests][BREAK] arguments are not filtered (since version 0.6.20 probably) 2022-09-05 10:12:47 +02:00
Maxime Alves LIRMM@home
5e21d4c24f fix ret_type feature 2022-08-31 00:14:10 +02:00
Maxime Alves LIRMM@home
8bdf5cab82 [testing] MODULE and CONFIG attributes 2022-08-31 00:13:29 +02:00
Maxime Alves LIRMM
910e1e1497 [release] 0.6.22-rc0 2022-08-18 20:24:57 +02:00
Maxime Alves LIRMM
53a691d985 [responses] html and plaintext return types as ret_type argument 2022-08-18 20:22:14 +02:00
Maxime Alves LIRMM@home
5d1b88daca [release] 0.6.21 2022-08-08 20:26:31 +02:00
Maxime Alves LIRMM@home
c0bd6ddc43 [release] 0.6.21 2022-08-08 20:21:05 +02:00
Maxime Alves LIRMM
779dd2d519 [deps] lock pipenv update 2022-07-18 20:35:49 +02:00
Maxime Alves LIRMM
84747e3f73 [release] 0.6.20-rc0 2022-07-18 20:34:12 +02:00
Maxime Alves LIRMM
18748808c9 [fix] argument handling 2022-07-18 18:47:38 +02:00
Maxime Alves LIRMM
34ede09fe2 [release] 0.6.19 2022-07-06 08:34:11 +02:00
Maxime Alves LIRMM
6ebdf765bd [deps] pytest v7, [ci/cd] no tests on python3.6 2022-07-06 08:29:29 +02:00
Maxime Alves LIRMM
2d47789f61 [ci/cd] parallel tests on multiple python versions 2022-07-06 08:15:49 +02:00
3ec6d7514e [release] 0.6.18 2022-06-23 15:48:58 +02:00
3c97d39cdc [config] restore dumped configuration when using HALFAPI_DOMAIN_NAME 2022-06-23 15:33:19 +02:00
f96c712aa8 [release] 0.6.17 2022-06-23 11:07:41 +02:00
23a93026aa [halfdomain] fix feature HALFAPI_DOMAIN_MODULE for acls route 2022-06-23 11:03:52 +02:00
05cf37c775 [deps] ajout virtualenv dans les deps (pour build) 2022-06-23 11:00:33 +02:00
a46b045ca6 [release] 0.6.16 2022-06-23 07:24:05 +02:00
f68b7e59b8 handle HALFAPI_DOMAIN_MODULE environment variable 2022-06-23 07:21:49 +02:00
Maxime Alves LIRMM@home
b17ce623f4 [release] 0.6.15 2022-05-17 16:19:41 +02:00
Maxime Alves LIRMM@home
7dd7d00625 [halfdomain] fix last modification 2022-05-17 16:13:33 +02:00
Maxime Alves LIRMM@home
389823db82 [deps] Pipfile lock 2022-05-17 16:02:23 +02:00
Maxime Alves LIRMM@home
e70239433f [halfdomain] add ability to specify a domain's acl module path in __acl__ attribute 2022-05-17 16:01:58 +02:00
Maxime Alves LIRMM@home
739ffc9afa [release] 0.6.14 2022-05-17 10:31:34 +02:00
Maxime Alves LIRMM@home
99d4aaeb8d [lib/responses] ajout format XLSX, car ODS bug avec les dates
https://github.com/pyexcel/pyexcel-ods/issues/31
2022-05-17 10:30:15 +02:00
Maxime Alves LIRMM
0d9dc2a018 [release] 0.6.13
- (rollback from 0.6.12) Remove pytest from dependencies in Docker file and
  remove tests
- (dep) Add "packaging" dependency
- Add dependency check capability when instantiating a domain (__deps__
  variable, see in dummy_domain)
2022-03-21 09:51:04 +01:00
Maxime Alves LIRMM
b63b0f52c6 [domain] checks versions specified in __deps__ variable of domain module 2022-03-21 09:45:35 +01:00
Maxime Alves LIRMM
63d6d1e8ea [release] 0.6.12 2022-03-17 17:48:41 +01:00
Maxime Alves LIRMM
e8c99e6012 [ci/cd] install pytest in docker image, run pytest at launch 2022-03-17 17:38:20 +01:00
Maxime Alves LIRMM
568aea9ea8 [clena] nettoyage des fonctions non utilisées 2022-03-16 13:09:50 +01:00
Maxime Alves LIRMM
988a1e5bab [ci/cd] fix kaniko call 2022-03-14 15:24:57 +01:00
Maxime Alves LIRMM
de72e469d2 [release] 0.6.11 2022-03-14 15:16:06 +01:00
Maxime Alves LIRMM
f7879c6388 [release] 0.6.10
- Add "x-out" field in HTTP headers when "out" parameters are specified for a
  route
- Add "out" kwarg for not-async functions that specify it

- Hide data fields in args_check logs

- Fix testing lib for domains (add default secret and debug option)

- Domains now need to include the following variables in their __init__.py
    - __name__ (str, optional)
    - __id__ (str, optional)
- halfapi domain

- Mounts domain routers with their ACLs as decorator
- Configuration example files for systemd and a system-wide halfapi install
- Runs projects
- Handles JWT authentication middleware
2022-03-08 19:24:24 +01:00
Maxime Alves LIRMM
84179743a6 [release] 0.6.9 2022-03-02 16:46:32 +01:00
Maxime Alves LIRMM
adf7f872b6 changelog 2022-02-28 10:21:01 +01:00
Maxime Alves LIRMM
b96f4908c6 [release] 0.6.8 2022-02-28 10:10:06 +01:00
Maxime Alves LIRMM
a388faf1d8 [testing] add "secret" and "production" variable to test_domain configuration 2022-02-28 09:58:47 +01:00
Maxime Alves LIRMM
4d6e935813 [ci/cd] fix kaniko executor 2022-02-10 16:31:26 +01:00
Maxime Alves LIRMM
4e080f805f [fix] odsresponse 2022-02-10 16:30:08 +01:00
Maxime Alves LIRMM
91dd4cbaa8 [ci/cd] check tag and not branch master 2022-02-10 16:01:38 +01:00
Maxime Alves LIRMM
209c6ef40a [ci/cd] add rules to execute build stage 2022-02-10 14:31:30 +00:00
Maxime Alves LIRMM
a1c1bf04df [ci/cd] stages 2022-02-10 15:15:08 +01:00
Maxime Alves LIRMM
90203b2edf [ci] add ./tests to pytest command 2022-02-10 12:38:54 +01:00
Maxime Alves LIRMM
a69d2b7639 [tests] async router renamed to async_router 2022-02-10 12:30:49 +01:00
Maxime Alves LIRMM
c3153921f7 [tests] test_domain import dummy_domain inside test class 2022-02-10 11:59:40 +01:00
Maxime Alves LIRMM
97fee8ca96 [ci] execute pytest on tests folder 2022-02-10 11:56:35 +01:00
62 changed files with 3028 additions and 1046 deletions

View File

@ -6,7 +6,7 @@
# Official language image. Look for the different tagged releases at:
# https://hub.docker.com/r/library/python/tags/
image: python:3.9-bullseye
image: python:alpine3.18
# Change pip's cache directory to be inside the project directory since we can
# only cache local items.
@ -27,33 +27,64 @@ cache:
- .cache/pip
- venv/
test:
script:
- apt-get update && apt-get -y install python3-venv
- python3 -V # Print out python version for debugging
- pip3 install pipenv
- pipenv install --dev --skip-lock
- pipenv run pytest --version
- pipenv run pytest -v
- pipenv run halfapi --version
stages:
- test
- build
run:
script:
- apt-get update && apt-get -y install python3-venv
.before_script_template: &test
before_script:
- python3 -V # Print out python version for debugging
- pip3 install pipenv
- pipenv install --dev --skip-lock
- pipenv run python -m build --sdist
- pipenv run python -m build --wheel
- python3 -m venv /tmp/venv
- /tmp/venv/bin/pip3 install .["tests","pyexcel"]
- /tmp/venv/bin/pip3 install coverage pytest
test:
image: python:alpine${ALPINEVERSION}
stage: test
<<: *test
parallel:
matrix:
- ALPINEVERSION: ["3.16", "3.17", "3.18", "3.19"]
script:
- /tmp/venv/bin/pytest --version
- PYTHONPATH=./tests/ /tmp/venv/bin/coverage run --source halfapi -m pytest
- /tmp/venv/bin/coverage xml
- /tmp/venv/bin/halfapi --version
artifacts:
reports:
coverage_report:
coverage_format: cobertura
path: coverage.xml
build_pypi:
stage: build
script:
- python3 -V # Print out python version for debugging
- python3 -m venv /tmp/venv
- /tmp/venv/bin/pip3 install .
artifacts:
paths:
- dist/*.whl
rules:
- if: '$CI_COMMIT_TAG != ""'
variables:
TAG: $CI_COMMIT_TAG
build:
build_container:
rules:
- if: '$CI_COMMIT_TAG != ""'
variables:
IMGTAG: $CI_COMMIT_TAG
- if: '$CI_COMMIT_REF_NAME == "devel"'
variables:
IMGTAG: "latest"
stage: build
image: $CI_REGISTRY/devtools/kaniko
script:
- echo "Will upload image halfapi:\"$IMGTAG\""
- echo "{\"auths\":{\"$CI_REGISTRY\":{\"username\":\"$CI_REGISTRY_USER\",\"password\":\"$CI_REGISTRY_PASSWORD\"}}}" > /kaniko/.docker/config.json
- /kaniko/executor --context $CI_PROJECT_DIR --dockerfile $CI_PROJECT_DIR/Dockerfile --destination $CI_REGISTRY/malves/halfapi:latest
- /kaniko/executor --force --context $CI_PROJECT_DIR --dockerfile $CI_PROJECT_DIR/Dockerfile --destination $CI_REGISTRY/malves/halfapi:"$IMGTAG"
artifacts:
paths:
- /kaniko/.docker/config.json

View File

@ -1,5 +1,237 @@
# HalfAPI
## 0.6.31
Dependencies updates
- orjson v3.10
- starlette v0.46.2
- schema v0.7.7
- pyjwt v2.10.0
Warning : the on_startup halfAPI argument is now removed, use the lifeSpan
## 0.6.30
Dependencies updates
- pyYAML v6.0.1
- starlette v0.37.2
Warning : the on_startup halfAPI argument is now removed, use the lifeSpan
## 0.6.29
### Dependencies
Starlette version bumped to 0.33.
## 0.6.28
### Dependencies
Starlette version bumped to 0.31 (had to disable a test to make it work but
seems not important).
### Development dependencies
Python 3.7 is no longer supported (openapi_spec_validator is not compatible).
If you are a developper, you should update dev dependencies in your virtual
environment.
### OpenAPI schemas
This release improves OpenAPI specification in routes, and gives a default
"parameters" field for routes that have path parameters.
Also, if you use halfAPI for multi-domain setups, you may be annoyed by the
change in the return value of the "/" route that differs from "/domain" route.
An HalfAPI instance should return one and only one OpenAPI Schema, so you can
rely on it to connect to other software.
The version number that is contained under the "info" dictionnary is now the "version"
of the Api domain, as specified in the domain dictionnary specified at the root
of the Domain.
The title field of the "info" dictionnary is now the Domain's name.
The ACLs list is now available under the "info.x-acls" attribute of the schema.
It is still accessible from the "/halfapi/acls" route.
#### Schema Components
You can now specify a dict in the domain's metadata dict that follows the
"components" key of an OpenAPI schema.
Use it to define models that are used in your API. You can see an exemple of
it's use in the "tests/dummy_domain/__init__.py" file.
### ACLs
The use of an "HEAD" request to check an ACL is now the norm. Please change all
the occurrences of your calls on theses routes with the GET method.
### CLI
Domain command update :
The `--conftest` flag is now allowed when running the `domain` command, it dumps the current configuration as a TOML string.
`halfapi domain --conftest my_domain`
The `--dry-run` flag was buggy and is now fixed when using the `domai ` command with the `--run` flag.
### Configuration
The `port` option in a `domain.my_domain` section in the TOML config file is now prefered to the one in the `project` section.
The `project` section is used as a default section for the whole configuration file. - Tests still have to be written -
The standard configuration precedence is fixed, in this order from the hight to the lower :
- Argument value (i.e. : --log-level)
- Environment value (i.e. : HALFAPI_LOGLEVEL)
- Configuration value under "domain" key
- Configuration value under "project" key
- Default configuration value given in the "DEFAULT_CONF" dictionary of halfapi/conf.py
### Logs
Small cleanup of the logs levels. If you don't want the config to be dumped, just set the HALFAPI_LOGLEVEL to something different than "DEBUG".
### Fixes
- Check an ACL based on a decorator on "/halfapi/acls/MY_ACL"
## 0.6.27
### Breaking changes
- ACLs definition can now include a "public" parameter that defines if there should be an automatic creation of a route to check this acls
- /halfapi/acls does not return the "result", it just returns if there is a public route to check the ACL on /halfapi/acls/acl_name
=======
argument of starlette instead.
>>>>>>> a8c59c6 ([release] halfapi 0.6.27)
## 0.6.26
- Adds the "base_url", "cookies" and "url" to the "halfapi" argument of route definitions
## 0.6.25
- Deletes the "Authorization" cookie on authentication error
- Do not raise an exception on signature expiration, use "Nobody" user instead
## 0.6.24
- Uses the "Authorization" cookie to read authentication token additionnaly to the "Authorization" header
- CLI : allows to run a single domain using the "halfapi domain --run domain_name" command
## 0.6.23
Dependency update version
- starlette v0.23
- orjson v3.8.5
- click v8
- pyJWT v2.6
- pyYAML v6
- toml v0.10
## 0.6.22
- IMPORTANT : Fix bug introduced with 0.6.20 (fix arguments handling)
- BREAKING : A domain should now include it's meta datas in a "domain" dictionary located in the __init__.py file of the domain's root. Consider looking in 'tests/dummy_domain/__init__.py'
- Add *html* return type as default argument ret_type
- Add *txt* return type
- Log unhandled exceptions
- Log HTTPException with statuscode 500 as critical
- PyJWT >=2.4.0,<2.5.0
## 0.6.21
- Store only domain's config in halfapi['config']
- Should run halfapi domain with config_file argument
- Testing : You can specify a "MODULE" attribute to point out the path to the Api's base module
- Testing : You can specify a "CONFIG" attribute to set the domain's testing configuration
- Environment : HALFAPI_DOMAIN_MODULE can be set to specify Api's base module
- Config : 'module' attribute can be set to specify Api's base module
## 0.6.20
- Fix arguments handling
## 0.6.19
- Allow file sending in multipart request (#32)
- Add python-multipart dependency
## 0.6.18
- Fix config coming from .halfapi/config when using HALFAPI_DOMAIN_NAME environment variable
## 0.6.17
- Fix 0.6.16
- Errata : HALFAPI_DOMAIN is HALFAPI_DOMAIN_NAME
- Testing : You can now specify "MODULE" class attribute for your "test_domain"
subclasses
## 0.6.16
- The definition of "HALFAPI_DOMAIN_MODULE" environment variable allows to
specify the base module for a domain structure. It is formatted as a python
import path.
The "HALFAPI_DOMAIN" specifies the "name" of the module
## 0.6.15
- Allows to define a "__acl__" variable in the API module's __init__.py, to
specify how to import the acl lib. You can also specify "acl" in the domain's
config
## 0.6.14
- Add XLSXResponse (with format argument set to "xlsx"), to return .xlsx files
## 0.6.13
- (rollback from 0.6.12) Remove pytest from dependencies in Docker file and
remove tests
- (dep) Add "packaging" dependency
- Add dependency check capability when instantiating a domain (__deps__
variable, see in dummy_domain)
## 0.6.12
- Installs pytest with dependencies in Docker image, tests when building image
## 0.6.11
- Fix "request" has no "path_params" attribute bug
## 0.6.10
- Add "x-out" field in HTTP headers when "out" parameters are specified for a
route
- Add "out" kwarg for not-async functions that specify it
## 0.6.9
- Hide data fields in args_check logs
## 0.6.8
- Fix testing lib for domains (add default secret and debug option)
## 0.6.2
- Domains now need to include the following variables in their __init__.py

View File

@ -1,9 +1,11 @@
# syntax=docker/dockerfile:1
FROM docker.io/python:3.8.12-slim-bullseye
FROM python:alpine3.19
COPY . /halfapi
WORKDIR /halfapi
RUN apt-get update > /dev/null && apt-get -y install git > /dev/null
RUN pip install gunicorn uvicorn
RUN pip install .
CMD gunicorn halfapi.app
ENV VENV_DIR=/opt/venv
RUN mkdir -p $VENV_DIR
RUN python -m venv $VENV_DIR
RUN $VENV_DIR/bin/pip install gunicorn uvicorn
RUN $VENV_DIR/bin/pip install .
RUN ln -s $VENV_DIR/bin/halfapi /usr/local/bin/
CMD $VENV_DIR/bin/gunicorn halfapi.app

26
Pipfile
View File

@ -4,25 +4,33 @@ url = "https://pypi.org/simple"
verify_ssl = true
[dev-packages]
pytest = "*"
pytest = ">=7,<8"
requests = "*"
pytest-asyncio = "*"
pylint = "*"
build = "*"
pytest-pythonpath = "*"
twine = "*"
pyflakes = "*"
vulture = "*"
virtualenv = "*"
httpx = "*"
openapi-schema-validator = "*"
openapi-spec-validator = "*"
coverage = "*"
[packages]
click = ">=7.1,<8"
starlette = ">=0.17,<0.18"
click = ">=8,<9"
starlette = ">=0.46,<0.47"
uvicorn = ">=0.13,<1"
orjson = ">=3.4.7,<4"
pyjwt = ">=2.3.0,<2.4.0"
pyyaml = ">=5.3.1,<6"
orjson = ">=3.10,<4"
pyjwt = ">=2.10.0,<2.11.0"
pyyaml = ">=6.0.1,<7"
timing-asgi = ">=0.2.1,<1"
schema = ">=0.7.4,<1"
toml = "*"
schema = ">=0.7.7,<1"
toml = ">=0.10,<0.11"
pip = "*"
packaging = ">=19.0"
python-multipart = "*"
[scripts]
halfapi = "python -m halfapi"

1435
Pipfile.lock generated

File diff suppressed because it is too large Load Diff

View File

@ -23,17 +23,15 @@ to reference [HalfORM](https://github.com/collorg/halfORM), a project written by
Configure HalfAPI in the file : .halfapi/config .
It's an **ini** file that contains at least two sections, project and domains.
It's a **toml** file that contains at least two sections, project and domains.
https://toml.io/en/
### Project
The main configuration options without which HalfAPI cannot be run.
**name** : Project's name
**halfapi_version** : The HalfAPI version on which you work
**secret** : The file containing the secret to decode the user's tokens.
**port** : The port for the test server.
@ -43,12 +41,28 @@ The main configuration options without which HalfAPI cannot be run.
### Domains
The name of the options should be the name of the domains' module, the value is the
submodule which contains the routers.
Specify the domains configurations in the following form :
Example :
```
[domains.DOMAIN_NAME]
name = "DOMAIN_NAME"
enabled = true
prefix = "/prefix"
module = "domain_name.path.to.api.root"
port = 1002
```
dummy_domain = .routers
Specific configuration can be done under the "config" section :
```
[domains.DOMAIN_NAME.config]
boolean_option = false
string_value = "String"
answer = 42
listylist = ["hello", "world"]
```
And can be accessed through the app's "config" dictionnary.
## Usage
@ -62,9 +76,17 @@ Run the project by using the `halfapi run` command.
You can try the dummy_domain with the following command.
```
python -m halfapi routes --export --noheader dummy_domain.routers | python -m halfapi run -
PYTHONPATH=$PWD/tests python -m halfapi domain dummy_domain
```
### CLI documentation
Use the CLI help.
```
python -m halfapi --help
python -m halfapi domain --help
```
## API Testing

View File

@ -1,5 +1,5 @@
#!/usr/bin/env python3
__version__ = '0.6.7'
__version__ = '0.6.31'
def version():
return f'HalfAPI version:{__version__}'

View File

@ -1,9 +1,7 @@
import os
from .halfapi import HalfAPI
from .conf import CONFIG, SCHEMA
from .logging import logger
logger.info('CONFIG: %s', CONFIG)
logger.info('SCHEMA: %s', SCHEMA)
application = HalfAPI(
CONFIG, SCHEMA or None).application
def application():
from .conf import CONFIG
return HalfAPI(CONFIG).application

View File

@ -8,19 +8,22 @@ import sys
import importlib
import subprocess
import json
import toml
import click
import orjson
import uvicorn
from .cli import cli
from ..conf import write_config, CONFIG
from ..conf import CONFIG
from ..half_domain import HalfDomain
from ..lib.domain import domain_schema
from ..lib.schemas import schema_dict_dom
from ..lib.routes import api_routes
from ..lib.responses import ORJSONResponse
from ..conf import CONFIG, PROJECT_LEVEL_KEYS
from ..logging import logger
@ -118,13 +121,22 @@ def list_api_routes():
# list_routes(domain, m_dom)
@click.option('--read',default=True, is_flag=True)
@click.option('--devel',default=None, is_flag=True)
@click.option('--watch',default=False, is_flag=True)
@click.option('--production',default=None, is_flag=True)
@click.option('--port',default=None, type=int)
@click.option('--log-level',default=None, type=str)
@click.option('--dry-run',default=False, is_flag=True)
@click.option('--run',default=False, is_flag=True)
@click.option('--read',default=False, is_flag=True)
@click.option('--conftest',default=False, is_flag=True)
@click.option('--create',default=False, is_flag=True)
@click.option('--update',default=False, is_flag=True)
@click.option('--delete',default=False, is_flag=True)
@click.argument('config_file', type=click.File(mode='rb'), required=False)
@click.argument('domain',default=None, required=False)
@cli.command()
def domain(domain, delete, update, create, read): #, domains, read, create, update, delete):
def domain(domain, config_file, delete, update, create, conftest, read, run, dry_run, log_level, port, production, watch, devel):
"""
The "halfapi domain" command
@ -141,27 +153,90 @@ def domain(domain, delete, update, create, read): #, domains, read, create, upd
# TODO: Connect to the create_domain function
raise NotImplementedError
raise Exception('Missing domain name')
if update:
if config_file:
ARG_CONFIG = toml.load(config_file.name)
if 'project' in ARG_CONFIG:
for key, value in ARG_CONFIG['project'].items():
if key in PROJECT_LEVEL_KEYS:
CONFIG[key] = value
if 'domain' in ARG_CONFIG and domain in ARG_CONFIG['domain']:
for key, value in ARG_CONFIG['domain'][domain].items():
if key in PROJECT_LEVEL_KEYS:
CONFIG[key] = value
CONFIG['domain'].update(ARG_CONFIG['domain'])
if create:
raise NotImplementedError
if delete:
elif update:
raise NotImplementedError
if read:
from ..conf import CONFIG
elif delete:
raise NotImplementedError
elif read:
from ..halfapi import HalfAPI
try:
config_domain = CONFIG.pop('domain').get(domain, {})
except KeyError:
config_domain = {}
halfapi = HalfAPI(CONFIG)
half_domain = halfapi.add_domain(domain, config=config_domain)
click.echo(orjson.dumps(
half_domain.schema(),
halfapi.domains[domain].schema(),
option=orjson.OPT_NON_STR_KEYS,
default=ORJSONResponse.default_cast)
)
else:
if dry_run:
CONFIG['dryrun'] = True
domains = CONFIG.get('domain')
for key in domains.keys():
if key != domain:
domains[key]['enabled'] = False
else:
domains[key]['enabled'] = True
if not log_level:
log_level = CONFIG.get('domain', {}).get('loglevel', CONFIG.get('loglevel', False))
else:
CONFIG['loglevel'] = log_level
if not port:
port = CONFIG.get('domain', {}).get('port', CONFIG.get('port', False))
else:
CONFIG['port'] = port
if devel is None and production is not None and (production is False or production is True):
CONFIG['production'] = production
if devel is not None:
CONFIG['production'] = False
CONFIG['loglevel'] = 'debug'
if conftest:
click.echo(
toml.dumps(CONFIG)
)
else:
# domain section port is preferred, if it doesn't exist we use the global one
uvicorn_kwargs = {}
if CONFIG.get('port'):
uvicorn_kwargs['port'] = CONFIG['port']
if CONFIG.get('loglevel'):
uvicorn_kwargs['log_level'] = CONFIG['loglevel'].lower()
if watch:
uvicorn_kwargs['reload'] = True
uvicorn.run(
'halfapi.app:application',
factory=True,
**uvicorn_kwargs
)
sys.exit(0)

View File

@ -17,8 +17,6 @@ from ..logging import logger
# from ..lib.domain import domain_schema_dict
from ..lib.constants import DOMAIN_SCHEMA, ROUTE_SCHEMA
from ..lib.responses import ORJSONResponse
# from ..lib.routes import api_routes
from ..lib.schemas import schema_to_csv # get_api_routes
@click.argument('module', required=True)
@click.option('--export', default=False, is_flag=True)

View File

@ -36,7 +36,7 @@ It follows the following format :
"""
import logging
from .logging import logger
import os
from os import environ
import sys
@ -46,22 +46,51 @@ import uuid
import toml
from .lib.domain import d_domains
from .logging import logger
PRODUCTION = True
LOGLEVEL = 'info'
CONF_FILE = os.environ.get('HALFAPI_CONF_FILE', '.halfapi/config')
DRYRUN = bool(os.environ.get('HALFAPI_DRYRUN', False))
SCHEMA = {}
CONF_DIR = environ.get('HALFAPI_CONF_DIR', '/etc/half_api')
DEFAULT_CONF = {
# Default configuration values
'SECRET': tempfile.mkstemp()[1],
'PROJECT_NAME': os.getcwd().split('/')[-1],
'PRODUCTION': True,
'HOST': '127.0.0.1',
'PORT': 3000,
'LOGLEVEL': 'info',
'BASE_DIR': os.getcwd(),
'CONF_FILE': '.halfapi/config',
'CONF_DIR': '/etc/half_api',
'DRYRUN': None
}
PROJECT_LEVEL_KEYS = {
# Allowed keys in "project" section of configuration file
'project_name',
'production',
'secret',
'host',
'port',
'loglevel',
'dryrun'
}
DOMAIN_LEVEL_KEYS = PROJECT_LEVEL_KEYS | {
# Allowed keys in "domain" section of configuration file
'name',
'module',
'prefix',
'enabled'
}
CONF_FILE = os.environ.get('HALFAPI_CONF_FILE', DEFAULT_CONF['CONF_FILE'])
CONF_DIR = os.environ.get('HALFAPI_CONF_DIR', DEFAULT_CONF['CONF_DIR'])
HALFAPI_ETC_FILE=os.path.join(
CONF_DIR, 'config'
)
BASE_DIR = os.environ.get('HALFAPI_BASE_DIR', DEFAULT_CONF['BASE_DIR'])
HALFAPI_DOT_FILE=os.path.join(
os.getcwd(), '.halfapi', 'config')
BASE_DIR, '.halfapi', 'config')
HALFAPI_CONFIG_FILES = []
@ -69,102 +98,108 @@ try:
with open(HALFAPI_ETC_FILE, 'r'):
HALFAPI_CONFIG_FILES.append(HALFAPI_ETC_FILE)
except FileNotFoundError:
logger.error('Cannot find a configuration file under %s', HALFAPI_DOT_FILE)
logger.info('Cannot find a configuration file under %s', HALFAPI_ETC_FILE)
try:
with open(HALFAPI_DOT_FILE, 'r'):
HALFAPI_CONFIG_FILES.append(HALFAPI_DOT_FILE)
except FileNotFoundError:
logger.error('Cannot find a configuration file under %s', HALFAPI_DOT_FILE)
def conf_files():
return [
os.path.join(
CONF_DIR, 'default.ini'
),
os.path.join(
os.getcwd(), '.halfapi', 'config')]
logger.info('Cannot find a configuration file under %s', HALFAPI_DOT_FILE)
def write_config():
"""
Writes the current config to the highest priority config file
"""
# with open(conf_files()[-1], 'w') as halfapi_config:
# config.write(halfapi_config)
pass
ENVIRONMENT = {}
# Load environment variables allowed in configuration
if 'HALFAPI_DRYRUN' in os.environ:
ENVIRONMENT['dryrun'] = True
def read_config():
if 'HALFAPI_PROD' in os.environ:
ENVIRONMENT['production'] = bool(os.environ.get('HALFAPI_PROD'))
if 'HALFAPI_LOGLEVEL' in os.environ:
ENVIRONMENT['loglevel'] = os.environ.get('HALFAPI_LOGLEVEL').lower()
if 'HALFAPI_SECRET' in os.environ:
ENVIRONMENT['secret'] = os.environ.get('HALFAPI_SECRET')
if 'HALFAPI_HOST' in os.environ:
ENVIRONMENT['host'] = os.environ.get('HALFAPI_HOST')
if 'HALFAPI_PORT' in os.environ:
ENVIRONMENT['port'] = int(os.environ.get('HALFAPI_PORT'))
def read_config(filenames=HALFAPI_CONFIG_FILES):
"""
The highest index in "filenames" are the highest priorty
"""
d_res = {}
logger.info('Reading config files %s', HALFAPI_CONFIG_FILES)
for CONF_FILE in HALFAPI_CONFIG_FILES:
d_res.update( toml.load(HALFAPI_CONFIG_FILES) )
logger.info('Reading config files %s', filenames)
for CONF_FILE in filenames:
if os.path.isfile(CONF_FILE):
conf_dict = toml.load(CONF_FILE)
d_res.update(conf_dict)
logger.info('Reading config files (result) %s', d_res)
logger.info('Read config files (result) %s', d_res)
return { **d_res.get('project', {}), 'domain': d_res.get('domain', {}) }
CONFIG = read_config()
CONFIG.update(**ENVIRONMENT)
PROJECT_NAME = CONFIG.get('project_name',
environ.get('HALFAPI_PROJECT_NAME', os.getcwd().split('/')[-1]))
os.environ.get('HALFAPI_PROJECT_NAME', DEFAULT_CONF['PROJECT_NAME']))
if os.environ.get('HALFAPI_DOMAIN_NAME'):
# Force enabled domain by environment variable
DOMAIN_NAME = os.environ.get('HALFAPI_DOMAIN_NAME')
if 'domain' in CONFIG and DOMAIN_NAME in CONFIG['domain'] \
and 'config' in CONFIG['domain'][DOMAIN_NAME]:
domain_config = CONFIG['domain'][DOMAIN_NAME]['config']
else:
domain_config = {}
if environ.get('HALFAPI_DOMAIN_NAME'):
DOMAIN_NAME = environ.get('HALFAPI_DOMAIN_NAME')
CONFIG['domain'] = {}
CONFIG['domain'][DOMAIN_NAME] = {
'enabled': True,
'name': DOMAIN_NAME,
'prefix': False
}
CONFIG['domain'][DOMAIN_NAME]['config'] = domain_config
if os.environ.get('HALFAPI_DOMAIN_MODULE'):
# Specify the pythonpath to import the specified domain (defaults to global)
dom_module = os.environ.get('HALFAPI_DOMAIN_MODULE')
CONFIG['domain'][DOMAIN_NAME]['module'] = dom_module
if len(CONFIG.get('domain', {}).keys()) == 0:
logger.info('No domains')
# Bind
HOST = CONFIG.get('host',
environ.get('HALFAPI_HOST', '127.0.0.1'))
PORT = int(CONFIG.get(
'port',
environ.get('HALFAPI_PORT', '3000')))
# Secret
SECRET = CONFIG.get(
'secret',
environ.get('HALFAPI_SECRET'))
if not SECRET:
if 'secret' not in CONFIG:
# TODO: Create a temporary secret
_, SECRET = tempfile.mkstemp()
with open(SECRET, 'w') as secret_file:
CONFIG['secret'] = DEFAULT_CONF['SECRET']
with open(CONFIG['secret'], 'w') as secret_file:
secret_file.write(str(uuid.uuid4()))
try:
with open(SECRET, 'r') as secret_file:
CONFIG['secret'] = SECRET.strip()
with open(CONFIG['secret'], 'r') as secret_file:
CONFIG['secret'] = CONFIG['secret'].strip()
except FileNotFoundError as exc:
logger.info('Running without secret file: %s', SECRET or 'no file specified')
logger.warning('Running without secret file: %s', CONFIG['secret'] or 'no file specified')
PRODUCTION = bool(CONFIG.get(
'production',
environ.get('HALFAPI_PROD', True)))
CONFIG.setdefault('project_name', DEFAULT_CONF['PROJECT_NAME'])
CONFIG.setdefault('production', DEFAULT_CONF['PRODUCTION'])
CONFIG.setdefault('host', DEFAULT_CONF['HOST'])
CONFIG.setdefault('port', DEFAULT_CONF['PORT'])
CONFIG.setdefault('loglevel', DEFAULT_CONF['LOGLEVEL'])
CONFIG.setdefault('dryrun', DEFAULT_CONF['DRYRUN'])
LOGLEVEL = CONFIG.get(
'loglevel',
environ.get('HALFAPI_LOGLEVEL', 'info')).lower()
BASE_DIR = CONFIG.get(
'base_dir',
environ.get('HALFAPI_BASE_DIR', '.'))
CONFIG['project_name'] = PROJECT_NAME
CONFIG['production'] = PRODUCTION
CONFIG['secret'] = SECRET
CONFIG['host'] = HOST
CONFIG['port'] = PORT
CONFIG['dryrun'] = DRYRUN
# !!!TO REMOVE!!!
SECRET = CONFIG.get('secret')
PRODUCTION = CONFIG.get('production')
# !!!

View File

@ -3,13 +3,20 @@ import inspect
import os
import re
from packaging.specifiers import SpecifierSet
from packaging.version import Version
from typing import Coroutine, Dict, Iterator, List, Tuple
from types import ModuleType, FunctionType
from schema import SchemaError
from starlette.applications import Starlette
from starlette.routing import Router
from starlette.middleware import Middleware
from starlette.routing import Router, Route
from starlette.schemas import SchemaGenerator
from .lib.acl import AclRoute
from .lib.responses import ORJSONResponse
import yaml
@ -17,15 +24,17 @@ import yaml
from . import __version__
from .lib.constants import API_SCHEMA_DICT, ROUTER_SCHEMA, VERBS
from .half_route import HalfRoute
from .lib import acl
from .lib import acl as lib_acl
from .lib.responses import PlainTextResponse
from .lib.routes import JSONRoute
from .lib.schemas import param_docstring_default
from .lib.domain import MissingAclError, PathError, UnknownPathParameterType, \
UndefinedRoute, UndefinedFunction, get_fct_name, route_decorator
from .lib.domain_middleware import DomainMiddleware
from .logging import logger
class HalfDomain(Starlette):
def __init__(self, domain, router=None, app=None):
def __init__(self, domain, module=None, router=None, acl=None, app=None):
"""
Parameters:
domain (str): Module name (should be importable)
@ -35,70 +44,191 @@ class HalfDomain(Starlette):
"""
self.app = app
self.m_domain = importlib.import_module(domain)
self.name = getattr(self.m_domain, '__name__', domain)
self.id = getattr(self.m_domain, '__id__')
self.version = getattr(self.m_domain, '__version__', '0.0.0')
# TODO: Check if given domain halfapi_version matches with __version__
self.halfapi_version = getattr(self.m_domain, '__halfapi_version__', __version__)
self.m_domain = importlib.import_module(domain) if module is None else module
self.d_domain = getattr(self.m_domain, 'domain', domain)
self.name = self.d_domain['name']
self.id = self.d_domain['id']
self.version = self.d_domain['version']
self.halfapi_version = self.d_domain.get('halfapi_version', __version__)
self.deps = self.d_domain.get('deps', tuple())
self.schema_components = self.d_domain.get('schema_components', dict())
if not router:
self.router = getattr('__router__', domain, '.routers')
self.router = self.d_domain.get('routers', '.routers')
else:
self.router = router
self.m_router = importlib.import_module(self.router, domain)
self.m_router = None
try:
self.m_router = importlib.import_module(self.router, self.m_domain.__package__)
except AttributeError:
raise Exception('no router module')
self.m_acl = importlib.import_module(f'{domain}.acl')
self.m_acl = HalfDomain.m_acl(self.m_domain, acl)
self.config = { **app.config }
logger.info('HalfDomain creation %s %s', domain, self.config)
for elt in self.deps:
package, version = elt
specifier = SpecifierSet(version)
package_module = importlib.import_module(package)
if Version(package_module.__version__) not in specifier:
raise Exception(
'Wrong version for package {} version {} (excepting {})'.format(
package, package_module.__version__, specifier
))
super().__init__(
routes=self.gen_domain_routes(),
middleware=[
(DomainMiddleware, {
'domain': {
Middleware(
DomainMiddleware,
domain={
'name': self.name,
'id': self.id,
'version': self.version,
'halfapi_version': self.halfapi_version,
'config': self.config.get('domain', {}).get(self.name, {}).get('config', {})
}
})
)
]
)
@staticmethod
def acls(domain):
""" Returns the ACLS constant for the given domain
def name(module):
""" Returns the name declared in the 'domain' dict at the root of the package
"""
m_acl = importlib.import_module(f'{domain}.acl')
try:
return getattr(m_acl, 'ACLS')
except AttributeError:
raise Exception(f'Missing acl.ACLS constant in {domain} module')
return module.domain['name']
@staticmethod
def acls_route(domain):
d_res = {}
m_acl = importlib.import_module(f'{domain}.acl')
for acl_name, doc, order in HalfDomain.acls(domain):
fct = getattr(m_acl, acl_name)
d_res[acl_name] = {
'callable': fct,
'docs': doc,
'result': None
def m_acl(module, acl=None):
""" Returns the imported acl module for the domain module
"""
if not acl:
acl = getattr(module, '__acl__', '.acl')
return importlib.import_module(acl, module.__package__)
@staticmethod
def acls(module, acl=None):
""" Returns the ACLS constant for the given domain
"""
m_acl = HalfDomain.m_acl(module, acl)
try:
return [
lib_acl.ACL(*elt)
for elt in getattr(m_acl, 'ACLS')
]
except AttributeError as exc:
logger.error(exc)
raise Exception(
f'Missing acl.ACLS constant in module {m_acl.__package__}') from exc
@staticmethod
def acls_route(domain, module_path=None, acl=None):
""" Dictionary of acls
Format :
{
[acl_name]: {
callable: fct_reference,
docs: fct_docstring,
}
}
"""
d_res = {}
module = importlib.import_module(domain) \
if module_path is None \
else importlib.import_module(module_path)
m_acl = HalfDomain.m_acl(module, acl)
for elt in HalfDomain.acls(module, acl=acl):
fct = getattr(m_acl, elt.name)
d_res[elt.name] = {
'callable': fct,
'docs': elt.documentation
}
return d_res
# def schema(self):
@staticmethod
def acls_router(domain, module_path=None, acl=None):
""" Returns a Router object with the following routes :
/ : The "acls" field of the API metadatas
/{acl_name} : If the ACL is defined as public, a route that returns either status code 200 or 401 on HEAD/GET request
"""
routes = []
d_res = {}
module = importlib.import_module(domain) \
if module_path is None \
else importlib.import_module(module_path)
m_acl = HalfDomain.m_acl(module, acl)
for elt in HalfDomain.acls(module, acl=acl):
fct = getattr(m_acl, elt.name)
d_res[elt.name] = {
'callable': fct,
'docs': elt.documentation,
'public': elt.public
}
if elt.public:
try:
if inspect.iscoroutinefunction(fct):
logger.warning('async decorator are not yet supported')
else:
inner = fct()
if inspect.iscoroutinefunction(fct) or callable(inner):
fct = inner
except TypeError:
# Fct is not a decorator or is not well called (has no default arguments)
# We can ignore this
pass
routes.append(
AclRoute(f'/{elt.name}', fct, elt)
)
d_res_under_domain_name = {}
d_res_under_domain_name[HalfDomain.name(module)] = d_res
routes.append(
Route(
'/',
JSONRoute(d_res_under_domain_name),
methods=['GET']
)
)
return Router(routes)
@staticmethod
def gen_routes(m_router: ModuleType,
verb: str,
path: List[str],
params: List[Dict]) -> Tuple[FunctionType, Dict]:
params: List[Dict],
path_param_docstrings: Dict[str, str] = {}) -> Tuple[FunctionType, Dict]:
"""
Returns a tuple of the function associatied to the verb and path arguments,
and the dictionary of it's acls
@ -130,18 +260,26 @@ class HalfDomain(Starlette):
fct_name = get_fct_name(verb, path[-1])
if hasattr(m_router, fct_name):
fct = getattr(m_router, fct_name)
fct_docstring_obj = yaml.safe_load(fct.__doc__)
if 'parameters' not in fct_docstring_obj and path_param_docstrings:
fct_docstring_obj['parameters'] = list(map(
yaml.safe_load,
path_param_docstrings.values()))
fct.__doc__ = yaml.dump(fct_docstring_obj)
else:
raise UndefinedFunction('{}.{}'.format(m_router.__name__, fct_name or ''))
if not inspect.iscoroutinefunction(fct):
return route_decorator(fct), params
else:
return acl.args_check(fct), params
# TODO: Remove when using only sync functions
return lib_acl.args_check(fct), params
@staticmethod
def gen_router_routes(m_router, path: List[str]) -> \
def gen_router_routes(m_router, path: List[str], PATH_PARAMS={}) -> \
Iterator[Tuple[str, str, ModuleType, Coroutine, List]]:
"""
Recursive generator that parses a router (or a subrouter)
@ -169,17 +307,32 @@ class HalfDomain(Starlette):
yield ('/'.join(filter(lambda x: len(x) > 0, path)),
verb,
m_router,
*HalfDomain.gen_routes(m_router, verb, path, params[verb])
*HalfDomain.gen_routes(m_router, verb, path, params[verb], PATH_PARAMS)
)
for subroute in params.get('SUBROUTES', []):
#logger.debug('Processing subroute **%s** - %s', subroute, m_router.__name__)
subroute_module = importlib.import_module(f'.{subroute}', m_router.__name__)
param_match = re.fullmatch('^([A-Z_]+)_([a-z]+)$', subroute)
parameter_name = None
if param_match is not None:
try:
parameter_name = param_match.groups()[0].lower()
if parameter_name in PATH_PARAMS:
raise Exception(f'Duplicate parameter name in same path! {subroute} : {parameter_name}')
parameter_type = param_match.groups()[1]
path.append('{{{}:{}}}'.format(
param_match.groups()[0].lower(),
param_match.groups()[1]))
parameter_name,
parameter_type,
)
)
try:
PATH_PARAMS[parameter_name] = subroute_module.param_docstring
except AttributeError as exc:
PATH_PARAMS[parameter_name] = param_docstring_default(parameter_name, parameter_type)
except AssertionError as exc:
raise UnknownPathParameterType(subroute) from exc
else:
@ -187,14 +340,19 @@ class HalfDomain(Starlette):
try:
yield from HalfDomain.gen_router_routes(
importlib.import_module(f'.{subroute}', m_router.__name__),
path)
subroute_module,
path,
PATH_PARAMS
)
except ImportError as exc:
logger.error('Failed to import subroute **{%s}**', subroute)
raise exc
path.pop()
if parameter_name:
PATH_PARAMS.pop(parameter_name)
path.pop()
@ -266,8 +424,8 @@ class HalfDomain(Starlette):
Generator(HalfRoute)
"""
yield HalfRoute('/',
JSONRoute([ self.schema() ]),
[{'acl': acl.public}],
self.schema_openapi(),
[{'acl': lib_acl.public}],
'GET'
)
@ -320,3 +478,36 @@ class HalfDomain(Starlette):
}
schema['paths'] = self.schema_dict()
return schema
def schema_openapi(self) -> Route:
schema = SchemaGenerator(
{
'openapi': '3.0.0',
'info': {
'title': self.name,
'version': self.version,
'x-acls': tuple(getattr(self.m_acl, 'ACLS', ())),
**({
f'x-{key}': value
for key, value in self.d_domain.items()
}),
},
'components': self.schema_components
}
)
async def inner(request, *args, **kwargs):
"""
description: |
Returns the current API routes description (OpenAPI v3)
as a JSON object
responses:
200:
description: API Schema in OpenAPI v3 format
"""
return ORJSONResponse(
schema.get_schema(routes=request.app.routes))
return inner

View File

@ -19,13 +19,13 @@ from .lib.domain import MissingAclError, PathError, UnknownPathParameterType, \
class HalfRoute(Route):
""" HalfRoute
"""
def __init__(self, path, fct, params, method):
def __init__(self, path: List[str], fct: Callable, params: List[Dict], method: str):
logger.info('HalfRoute creation: %s %s %s %s', path, fct, params, method)
if len(params) == 0:
raise MissingAclError('[{}] {}'.format(verb, '/'.join(path)))
raise MissingAclError('[{}] {}'.format(method, '/'.join(path)))
if len(path) == 0:
logger.error('Empty path for [{%s}]', verb)
logger.error('Empty path for [{%s}]', method)
raise PathError()
super().__init__(
@ -84,7 +84,11 @@ class HalfRoute(Route):
logger.debug(
'Args for current route (%s)', param.get('args'))
if 'out' in param:
req.scope['out'] = param['out']
if 'out' in param:
req.scope['out'] = param['out'].copy()
if 'check' in req.query_params:
return PlainTextResponse(param['acl'].__name__)

View File

@ -9,6 +9,7 @@ It defines the following globals :
- application (the asgi application itself - a starlette object)
"""
import sys
import logging
import time
import importlib
@ -19,7 +20,7 @@ from starlette.applications import Starlette
from starlette.authentication import UnauthenticatedUser
from starlette.exceptions import HTTPException
from starlette.middleware import Middleware
from starlette.routing import Route, Mount
from starlette.routing import Router, Route, Mount
from starlette.requests import Request
from starlette.responses import Response, PlainTextResponse
from starlette.middleware.authentication import AuthenticationMiddleware
@ -32,26 +33,36 @@ from timing_asgi.integrations import StarletteScopeToName
from .lib.constants import API_SCHEMA_DICT
from .lib.domain_middleware import DomainMiddleware
from .lib.timing import HTimingClient
from .lib.jwt_middleware import JWTAuthenticationBackend
from .lib.jwt_middleware import JWTAuthenticationBackend, on_auth_error
from .lib.responses import (ORJSONResponse, UnauthorizedResponse,
NotFoundResponse, InternalServerErrorResponse, NotImplementedResponse,
ServiceUnavailableResponse)
ServiceUnavailableResponse, gen_exception_route)
from .lib.domain import NoDomainsException
from .lib.routes import gen_schema_routes, JSONRoute
from .lib.schemas import schema_json, get_acls
from .lib.schemas import schema_json
from .logging import logger, config_logging
from .half_domain import HalfDomain
from halfapi import __version__
class HalfAPI(Starlette):
def __init__(self, config,
def __init__(self,
config,
d_routes=None):
config_logging(logging.DEBUG)
# Set log level (defaults to debug)
config_logging(
getattr(logging, config.get('loglevel', 'DEBUG').upper(), 'DEBUG')
)
self.config = config
SECRET = self.config.get('secret')
PRODUCTION = self.config.get('production', True)
DRYRUN = self.config.get('dryrun', False)
TIMINGMIDDLEWARE = self.config.get('timingmiddleware', False)
if DRYRUN:
logger.info('HalfAPI starting in dry-run mode')
else:
logger.info('HalfAPI starting')
self.PRODUCTION = PRODUCTION
self.SECRET = SECRET
@ -65,11 +76,15 @@ class HalfAPI(Starlette):
Mount('/halfapi', routes=list(self.halfapi_routes()))
)
logger.info('Config: %s', self.config)
logger.info('Active domains: %s',
list(filter(
lambda n: n.get('enabled', False),
self.config.get('domain', {}).values())))
logger.debug('Config: %s', self.config)
domains = {
key: elt
for key, elt in self.config.get('domain', {}).items()
if elt.get('enabled', False)
}
logger.debug('Active domains: %s', domains)
if d_routes:
# Mount the routes from the d_routes argument - domain-less mode
@ -90,20 +105,19 @@ class HalfAPI(Starlette):
debug=not PRODUCTION,
routes=routes,
exception_handlers={
401: UnauthorizedResponse,
404: NotFoundResponse,
500: InternalServerErrorResponse,
501: NotImplementedResponse,
503: ServiceUnavailableResponse
},
on_startup=startup_fcts
401: gen_exception_route(UnauthorizedResponse),
404: gen_exception_route(NotFoundResponse),
500: gen_exception_route(HalfAPI.exception),
501: gen_exception_route(NotImplementedResponse),
503: gen_exception_route(ServiceUnavailableResponse)
}
)
schemas = []
self.__domains = {}
for key, domain in self.config.get('domain', {}).items():
for key, domain in domains.items():
if not isinstance(domain, dict):
continue
@ -112,7 +126,7 @@ class HalfAPI(Starlette):
continue
if not domain.get('prefix', False):
if len(self.config.get('domain').keys()) > 1:
if len(domains.keys()) > 1:
raise Exception('Cannot use multiple domains and set prefix to false')
path = '/'
else:
@ -122,7 +136,12 @@ class HalfAPI(Starlette):
domain_key = domain.get('name', key)
self.add_domain(domain_key, domain.get('router'), path)
add_domain_args = {
**domain,
'path': path
}
self.add_domain(**add_domain_args)
schemas.append(self.__domains[domain_key].schema())
@ -132,10 +151,11 @@ class HalfAPI(Starlette):
if SECRET:
self.add_middleware(
AuthenticationMiddleware,
backend=JWTAuthenticationBackend()
backend=JWTAuthenticationBackend(),
on_error=on_auth_error
)
if not PRODUCTION:
if not PRODUCTION and TIMINGMIDDLEWARE:
self.add_middleware(
TimingMiddleware,
client=HTimingClient(),
@ -143,15 +163,24 @@ class HalfAPI(Starlette):
starlette_app=self)
)
@property
def version(self):
return __version__
async def version_async(self, request, *args, **kwargs):
"""
description: Version route
responses:
200:
description: Currently running HalfAPI's version
"""
return Response(self.version)
@staticmethod
async def exception(request: Request, exc: HTTPException):
logger.critical(exc, exc_info=True)
return InternalServerErrorResponse()
@property
def application(self):
return self
@ -161,11 +190,21 @@ class HalfAPI(Starlette):
"""
async def get_user(request, *args, **kwargs):
"""
description: WhoAmI route
responses:
200:
description: The currently logged-in user
content:
application/json:
schema:
type: object
"""
return ORJSONResponse({'user':request.user})
yield Route('/whoami', get_user)
yield Route('/schema', schema_json)
yield Route('/acls', self.acls_route())
yield Mount('/acls', self.acls_router())
yield Route('/version', self.version_async)
""" Halfapi debug routes definition
"""
@ -207,46 +246,71 @@ class HalfAPI(Starlette):
time.sleep(1)
sys.exit(0)
def acls_route(self):
res = {
domain: HalfDomain.acls_route(domain)
for domain, domain_conf in self.config.get('domain', {}).items()
if isinstance(domain_conf, dict) and domain_conf.get('enabled', False)
}
def acls_router(self):
mounts = {}
async def wrapped(req, *args, **kwargs):
for domain, domain_acls in res.items():
for acl_name, d_acl in domain_acls.items():
fct = d_acl['callable']
if not callable(fct):
raise Exception(
'No callable function in acl definition %s',
acl_name)
for domain, domain_conf in self.config.get('domain', {}).items():
if isinstance(domain_conf, dict) and domain_conf.get('enabled', False):
mounts['domain'] = HalfDomain.acls_router(
domain,
module_path=domain_conf.get('module'),
acl=domain_conf.get('acl')
)
fct_result = fct(req, *args, **kwargs)
if callable(fct_result):
fct_result = fct()(req, *args, **kwargs)
d_acl['result'] = fct_result
return ORJSONResponse(res)
return wrapped
if len(mounts) > 1:
return Router([
Mount(f'/{domain}', acls_router)
for domain, acls_router in mounts.items()
])
elif len(mounts) == 1:
return Mount('/', mounts.popitem()[1])
else:
return Router()
@property
def domains(self):
return self.__domains
def add_domain(self, name, router=None, path='/', config=None):
if config:
self.config['domain'][name] = config
def add_domain(self, **kwargs):
self.__domains[name] = HalfDomain(
name,
router,
self
)
if not kwargs.get('enabled'):
raise Exception(f'Domain not enabled ({kwargs})')
self.mount(path, self.__domains[name])
name = kwargs['name']
self.config['domain'][name] = kwargs.get('config', {})
if not kwargs.get('module'):
module = name
else:
module = kwargs.get('module')
try:
self.__domains[name] = HalfDomain(
name,
module=importlib.import_module(module),
router=kwargs.get('router'),
acl=kwargs.get('acl'),
app=self
)
except ImportError as exc:
print(
'Cannot instantiate HalfDomain {} with module {}'.format(
name,
module
))
raise exc
self.mount(kwargs.get('path', name), self.__domains[name])
return self.__domains[name]
def __main__():
return HalfAPI(CONFIG).application
if __name__ == '__main__':
__main__()

View File

@ -2,10 +2,14 @@
"""
Base ACL module that contains generic functions for domains ACL
"""
from dataclasses import dataclass
from functools import wraps
from json import JSONDecodeError
import yaml
from starlette.authentication import UnauthenticatedUser
from starlette.exceptions import HTTPException
from starlette.routing import Route
from starlette.responses import Response
from ..logging import logger
@ -28,7 +32,9 @@ def connected(fct=public):
or not hasattr(req.user, 'is_authenticated')):
return False
return fct(req, **{**kwargs, **req.path_params})
if hasattr(req, 'path_params'):
return fct(req, **{**kwargs, **req.path_params})
return fct(req, **{**kwargs})
return caller
@ -54,11 +60,23 @@ def args_check(fct):
data_ = dict(req.query_params)
elif req.method in ['POST', 'PATCH', 'PUT', 'DELETE']:
try:
data_ = await req.json()
except JSONDecodeError as exc:
logger.debug('Posted data was not JSON')
pass
if req.scope.get('headers'):
if b'content-type' not in dict(req.scope.get('headers')):
data_ = {}
else:
content_type = dict(req.scope.get('headers')).get(b'content-type').decode().split(';')[0]
if content_type == 'application/json':
try:
data_ = await req.json()
except JSONDecodeError as exc:
logger.debug('Posted data was not JSON')
pass
elif content_type in [
'multipart/form-data', 'application/x-www-form-urlencoded']:
data_ = dict(await req.form())
else:
data_ = await req.body()
def plural(array: list) -> str:
return 's' if len(array) > 1 else ''
@ -67,7 +85,7 @@ def args_check(fct):
args_d = req.scope.get('args')
if args_d is not None:
if args_d is not None and isinstance(data_, dict):
required = args_d.get('required', set())
missing = []
@ -85,16 +103,18 @@ def args_check(fct):
optional = args_d.get('optional', set())
for key in optional:
if key in data_:
data[key] = data_[key]
if key in data_:
data[key] = data_[key]
else:
""" Unsafe mode, without specified arguments
""" Unsafe mode, without specified arguments, or plain text mode
"""
data = data_
kwargs['data'] = data
logger.debug('args_check %s:%s %s %s', fct.__module__, fct.__name__, args, kwargs)
out_s = req.scope.get('out')
if out_s:
kwargs['out'] = list(out_s)
return await fct(req, *args, **kwargs)
@ -102,7 +122,57 @@ def args_check(fct):
# ACLS list for doc and priorities
# Write your own constant in your domain or import this one
# Format : (acl_name: str, acl_documentation: str, priority: int, [public=False])
#
# The 'priority' integer is greater than zero and the lower values means more
# priority. For a route, the order of declaration of the ACLs should respect
# their priority.
#
# When the 'public' boolean value is True, a route protected by this ACL is
# defined on the "/halfapi/acls/acl_name", that returns an empty response and
# the status code 200 or 401.
ACLS = (
('private', public.__doc__, 0),
('public', public.__doc__, 999)
('private', private.__doc__, 0, True),
('public', public.__doc__, 999, True)
)
@dataclass
class ACL():
name: str
documentation: str
priority: int
public: bool = False
class AclRoute(Route):
def __init__(self, path, acl_fct, acl: ACL):
self.acl_fct = acl_fct
self.name = acl.name
self.description = acl.documentation
self.docstring = yaml.dump({
'description': f'{self.name}: {self.description}',
'responses': {
'200': {
'description': 'ACL OK'
},
'401': {
'description': 'ACL FAIL'
}
}
})
async def endpoint(request, *args, **kwargs):
if request.method == 'GET':
logger.warning('Deprecated since 0.6.28, use HEAD method since now')
if self.acl_fct(request, *args, **kwargs) is True:
return Response(status_code=200)
return Response(status_code=401)
endpoint.__doc__ = self.docstring
return super().__init__(path, methods=['HEAD', 'GET'], endpoint=endpoint)

View File

@ -44,7 +44,7 @@ DOMAIN_SCHEMA = Schema({
Optional('version'): str,
Optional('patch_release'): str,
Optional('acls'): [
[str, str, int]
[str, str, int, Optional(bool)]
]
})

View File

@ -3,7 +3,6 @@
lib/domain.py The domain-scoped utility functions
"""
import os
import re
import sys
import importlib
@ -11,31 +10,41 @@ import inspect
from functools import wraps
from types import ModuleType, FunctionType
from typing import Coroutine, Generator
from typing import Dict, List, Tuple, Iterator
from typing import Dict, List, Tuple
import yaml
from starlette.exceptions import HTTPException
from halfapi.lib import acl
from halfapi.lib.responses import ORJSONResponse, ODSResponse
from halfapi.lib.responses import ORJSONResponse, ODSResponse, XLSXResponse, PlainTextResponse, HTMLResponse
# from halfapi.lib.router import read_router
from halfapi.lib.constants import VERBS
from ..logging import logger
class MissingAclError(Exception):
""" Exception to use when no acl are specified for a route
"""
pass
class PathError(Exception):
""" Exception to use when the path for a route is malformed
"""
pass
class UnknownPathParameterType(Exception):
""" Exception to use when the path parameter for a route is not supported
"""
pass
class UndefinedRoute(Exception):
""" Exception to use when the route definition cannot be found
"""
pass
class UndefinedFunction(Exception):
""" Exception to use when a function definition cannot be found
"""
pass
class NoDomainsException(Exception):
@ -43,13 +52,19 @@ class NoDomainsException(Exception):
"""
pass
def route_decorator(fct: FunctionType, ret_type: str = 'json') -> Coroutine:
def route_decorator(fct: FunctionType) -> Coroutine:
""" Returns an async function that can be mounted on a router
"""
@wraps(fct)
@acl.args_check
async def wrapped(request, *args, **kwargs):
fct_args_spec = inspect.getfullargspec(fct).args
fct_args_defaults = inspect.getfullargspec(fct).defaults or []
fct_args_defaults_dict = dict(list(zip(
reversed(fct_args_spec),
reversed(fct_args_defaults)
)))
fct_args = request.path_params.copy()
if 'halfapi' in fct_args_spec:
@ -58,32 +73,72 @@ def route_decorator(fct: FunctionType, ret_type: str = 'json') -> Coroutine:
'user' in request else None,
'config': request.scope.get('config', {}),
'domain': request.scope.get('domain', 'unknown'),
'cookies': request.cookies,
'base_url': request.base_url,
'url': request.url
}
if 'data' in fct_args_spec:
fct_args['data'] = kwargs.get('data')
if 'data' in fct_args_defaults_dict:
fct_args['data'] = fct_args_defaults_dict['data']
else:
fct_args['data'] = {}
""" If format argument is specified (either by get or by post param)
fct_args['data'].update(kwargs.get('data', {}))
if 'out' in fct_args_spec:
fct_args['out'] = kwargs.get('out')
""" If format argument is specified (either by get, post param or function argument)
"""
ret_type = fct_args.get('data', {}).get('format', 'json')
if 'ret_type' in fct_args_defaults_dict:
ret_type = fct_args_defaults_dict['ret_type']
else:
ret_type = fct_args.get('data', {}).get('format', 'json')
logger.debug('Return type {} (defaults: {})'.format(ret_type,
fct_args_defaults_dict))
try:
logger.debug('FCT_ARGS***** %s', fct_args)
if ret_type == 'json':
return ORJSONResponse(fct(**fct_args))
elif ret_type == 'ods':
if ret_type == 'ods':
res = fct(**fct_args)
assert isinstance(res, list)
for elt in res:
assert isinstance(elt, dict)
return ODSResponse(res)
else:
raise NotImplementedError
if ret_type == 'xlsx':
res = fct(**fct_args)
assert isinstance(res, list)
for elt in res:
assert isinstance(elt, dict)
return XLSXResponse(res)
if ret_type in ['html', 'xhtml']:
res = fct(**fct_args)
assert isinstance(res, str)
return HTMLResponse(res)
if ret_type in 'txt':
res = fct(**fct_args)
assert isinstance(res, str)
return PlainTextResponse(res)
raise NotImplementedError
except NotImplementedError as exc:
raise HTTPException(501) from exc
except Exception as exc:
# TODO: Write tests
logger.error(exc, exc_info=True)
if not isinstance(exc, HTTPException):
raise HTTPException(500) from exc
raise exc
@ -135,252 +190,3 @@ def get_fct_name(http_verb: str, path: str) -> str:
return '_'.join(fct_name)
def gen_routes(m_router: ModuleType,
verb: str,
path: List[str],
params: List[Dict]) -> Tuple[FunctionType, Dict]:
"""
Returns a tuple of the function associatied to the verb and path arguments,
and the dictionary of it's acls
Parameters:
- m_router (ModuleType): The module containing the function definition
- verb (str): The HTTP verb for the route (GET, POST, ...)
- path (List): The route path, as a list (each item being a level of
deepness), from the lowest level (domain) to the highest
- params (Dict): The acl list of the following format :
[{'acl': Function, 'args': {'required': [], 'optional': []}}]
Returns:
(Function, Dict): The destination function and the acl dictionary
"""
if len(params) == 0:
raise MissingAclError('[{}] {}'.format(verb, '/'.join(path)))
if len(path) == 0:
logger.error('Empty path for [{%s}]', verb)
raise PathError()
fct_name = get_fct_name(verb, path[-1])
if hasattr(m_router, fct_name):
fct = getattr(m_router, fct_name)
else:
raise UndefinedFunction('{}.{}'.format(m_router.__name__, fct_name or ''))
if not inspect.iscoroutinefunction(fct):
return route_decorator(fct), params
else:
return acl.args_check(fct), params
# def gen_router_routes(m_router: ModuleType, path: List[str]) -> \
#  Iterator[Tuple[str, str, ModuleType, Coroutine, List]]:
#  """
#  Recursive generator that parses a router (or a subrouter)
#  and yields from gen_routes
# 
#  Parameters:
# 
#  - m_router (ModuleType): The currently treated router module
#  - path (List[str]): The current path stack
# 
#  Yields:
# 
#  (str, str, ModuleType, Coroutine, List): A tuple containing the path, verb,
#  router module, function reference and parameters of the route.
#  Function and parameters are yielded from then gen_routes function,
#  that decorates the endpoint function.
#  """
# 
#  for subpath, params in read_router(m_router).items():
#  path.append(subpath)
# 
#  for verb in VERBS:
#  if verb not in params:
#  continue
#  yield ('/'.join(filter(lambda x: len(x) > 0, path)),
#  verb,
#  m_router,
#  *gen_routes(m_router, verb, path, params[verb])
#  )
# 
#  for subroute in params.get('SUBROUTES', []):
#  #logger.debug('Processing subroute **%s** - %s', subroute, m_router.__name__)
#  param_match = re.fullmatch('^([A-Z_]+)_([a-z]+)$', subroute)
#  if param_match is not None:
#  try:
#  path.append('{{{}:{}}}'.format(
#  param_match.groups()[0].lower(),
#  param_match.groups()[1]))
#  except AssertionError as exc:
#  raise UnknownPathParameterType(subroute) from exc
#  else:
#  path.append(subroute)
# 
#  try:
#  yield from gen_router_routes(
#  importlib.import_module(f'.{subroute}', m_router.__name__),
#  path)
# 
#  except ImportError as exc:
#  logger.error('Failed to import subroute **{%s}**', subroute)
#  raise exc
# 
#  path.pop()
# 
#  path.pop()
# 
# def domain_schema_dict(m_router: ModuleType) -> Dict:
#  """ gen_router_routes return values as a dict
#  Parameters:
# 
#  m_router (ModuleType): The domain routers' module
# 
#  Returns:
# 
#  Dict: Schema of dict is halfapi.lib.constants.DOMAIN_SCHEMA
# 
#  @TODO: Should be a "router_schema_dict" function
#  """
#  d_res = {}
# 
#  for path, verb, m_router, fct, parameters in gen_router_routes(m_router, []):
#  if path not in d_res:
#  d_res[path] = {}
# 
#  if verb not in d_res[path]:
#  d_res[path][verb] = {}
# 
#  d_res[path][verb]['callable'] = f'{m_router.__name__}:{fct.__name__}'
#  try:
#  d_res[path][verb]['docs'] = yaml.safe_load(fct.__doc__)
#  except AttributeError:
#  logger.error(
#  'Cannot read docstring from fct (fct=%s path=%s verb=%s', fct.__name__, path, verb)
# 
#  d_res[path][verb]['acls'] = list(map(lambda elt: { **elt, 'acl': elt['acl'].__name__ },
#  parameters))
# 
#  return d_res
from .constants import API_SCHEMA_DICT
def domain_schema(m_domain: ModuleType) -> Dict:
schema = { **API_SCHEMA_DICT }
routers_submod_str = getattr(m_domain, '__routers__', '.routers')
m_domain_acl = importlib.import_module('.acl', m_domain.__package__)
m_domain_routers = importlib.import_module(
routers_submod_str, m_domain.__package__
)
schema['domain'] = {
'name': getattr(m_domain, '__name__'),
'version': getattr(m_domain, '__version__', ''),
'patch_release': getattr(m_domain, '__patch_release__', ''),
'routers': routers_submod_str,
'acls': tuple(getattr(m_domain_acl, 'ACLS', ()))
}
schema['paths'] = domain_schema_dict(m_domain_routers)
return schema
def domain_schema_list(m_router: ModuleType) -> List:
""" Schema as list, one row by route/acl
Parameters:
m_router (ModuleType): The domain routers' module
Returns:
List[Tuple]: (path, verb, callable, doc, acls)
"""
res = []
for path, verb, m_router, fct, parameters in gen_router_routes(m_router, []):
for params in parameters:
res.append((
path,
verb,
f'{m_router.__name__}:{fct.__name__}',
params.get('acl').__name__,
params.get('args', {}).get('required', []),
params.get('args', {}).get('optional', []),
params.get('out', [])
))
return res
def d_domains(config) -> Dict[str, ModuleType]:
"""
Parameters:
config (ConfigParser): The .halfapi/config based configparser object
Returns:
dict[str, ModuleType]
"""
domains = {}
if os.environ.get('HALFAPI_DOMAIN_NAME') and os.environ.get('HALFAPI_DOMAIN_MODULE', '.routers'):
domains[os.environ.get('HALFAPI_DOMAIN_NAME')] = os.environ.get('HALFAPI_DOMAIN_MODULE')
elif 'domains' in config:
domains = dict(config['domains'].items())
try:
return {
domain: importlib.import_module(''.join((domain, module)))
for domain, module in domains.items()
}
except ImportError as exc:
logger.error('Could not load a domain : %s', exc)
raise exc
def router_acls(route_params: Dict, path: List, m_router: ModuleType) -> Generator:
for verb in VERBS:
params = route_params.get(verb)
if params is None:
continue
if len(params) == 0:
logger.error('No ACL for route [{%s}] %s', verb, "/".join(path))
else:
for param in params:
fct_acl = param.get('acl')
if not isinstance(fct_acl, FunctionType):
continue
yield fct_acl.__name__, fct_acl
def domain_acls(m_router, path):
routes = read_router(m_router)
for subpath, route_params in routes.items():
path.append(subpath)
yield from router_acls(route_params, path, m_router)
subroutes = route_params.get('SUBROUTES', [])
for subroute in subroutes:
logger.debug('Processing subroute **%s** - %s', subroute, m_router.__name__)
path.append(subroute)
try:
submod = importlib.import_module(f'.{subroute}', m_router.__name__)
except ImportError as exc:
logger.error('Failed to import subroute **{%s}**', subroute)
raise exc
yield from domain_acls(submod, path)
path.pop()
path.pop()

View File

@ -17,7 +17,7 @@ class DomainMiddleware(BaseHTTPMiddleware):
- acl
"""
def __init__(self, app, domain):
def __init__(self, app, domain=None):
""" app: HalfAPI instance
"""
logger.info('DomainMiddleware app:%s domain:%s', app, domain)
@ -35,7 +35,21 @@ class DomainMiddleware(BaseHTTPMiddleware):
request.scope['domain'] = self.domain['name']
if hasattr(request.app, 'config') \
and isinstance(request.app.config, dict):
request.scope['config'] = { **request.app.config }
# Set the config scope to the domain's config
request.scope['config'] = request.app.config.get(
'domain', {}
).get(
self.domain['name'], {}
).copy()
# TODO: Remove in 0.7.0
config = request.scope['config'].copy()
request.scope['config']['domain'] = {}
request.scope['config']['domain'][self.domain['name']] = {}
request.scope['config']['domain'][self.domain['name']]['config'] = config
else:
logger.debug('%s', request.app)
logger.debug('%s', getattr(request.app, 'config', None))
@ -56,6 +70,11 @@ class DomainMiddleware(BaseHTTPMiddleware):
response.headers['x-args-optional'] = \
','.join(request.scope['args']['optional'])
if len(request.scope.get('out', set())):
response.headers['x-out'] = \
','.join(request.scope['out'])
response.headers['x-domain'] = self.domain['name']
return response

View File

@ -14,16 +14,18 @@ from os import environ
import typing
from uuid import UUID
from http.cookies import SimpleCookie
import jwt
from starlette.authentication import (
AuthenticationBackend, AuthenticationError, BaseUser, AuthCredentials,
UnauthenticatedUser)
from starlette.requests import HTTPConnection
from starlette.requests import HTTPConnection, Request
from starlette.exceptions import HTTPException
from .user import CheckUser, JWTUser, Nobody
from ..logging import logger
from ..conf import CONFIG
from ..lib.responses import ORJSONResponse
SECRET=None
@ -34,6 +36,20 @@ except FileNotFoundError:
logger.error('Could not import SECRET variable from conf module,'\
' using HALFAPI_SECRET environment variable')
def cookies_from_scope(scope):
cookie = dict(scope.get("headers") or {}).get(b"cookie")
if not cookie:
return {}
simple_cookie = SimpleCookie()
simple_cookie.load(cookie.decode("utf8"))
return {key: morsel.value for key, morsel in simple_cookie.items()}
def on_auth_error(request: Request, exc: Exception):
response = ORJSONResponse({"error": str(exc)}, status_code=401)
response.delete_cookie('Authorization')
return response
class JWTAuthenticationBackend(AuthenticationBackend):
def __init__(self, secret_key: str = SECRET,
algorithm: str = 'HS256', prefix: str = 'JWT'):
@ -52,17 +68,22 @@ class JWTAuthenticationBackend(AuthenticationBackend):
self, conn: HTTPConnection
) -> typing.Optional[typing.Tuple['AuthCredentials', 'BaseUser']]:
# Standard way to authenticate via API
# https://datatracker.ietf.org/doc/html/rfc7235#section-4.2
token = conn.headers.get('Authorization')
if not token:
token = cookies_from_scope(conn.scope).get('Authorization')
is_check_call = 'check' in conn.query_params
is_fake_user_id = is_check_call and 'user_id' in conn.query_params
PRODUCTION = conn.scope['app'].debug == False
if not token and not is_check_call:
return AuthCredentials(), Nobody()
try:
if token and not is_fake_user_id:
if token:
payload = jwt.decode(token,
key=self.secret_key,
algorithms=[self.algorithm],
@ -71,24 +92,18 @@ class JWTAuthenticationBackend(AuthenticationBackend):
})
if is_check_call:
if is_fake_user_id:
try:
fake_user_id = UUID(conn.query_params['user_id'])
return AuthCredentials(), CheckUser(fake_user_id)
except ValueError as exc:
raise HTTPException(400, 'user_id parameter not an uuid')
if token:
return AuthCredentials(), CheckUser(payload['user_id'])
else:
return AuthCredentials(), Nobody()
return AuthCredentials(), Nobody()
if PRODUCTION and 'debug' in payload.keys() and payload['debug']:
raise AuthenticationError(
'Trying to connect using *DEBUG* token in *PRODUCTION* mode')
except jwt.ExpiredSignatureError as exc:
return AuthCredentials(), Nobody()
except jwt.InvalidTokenError as exc:
raise AuthenticationError(str(exc)) from exc
except Exception as exc:

View File

@ -56,10 +56,8 @@ def parse_query(q_string: str = ""):
>>> parse_query('limit:10')
<function parse_query.<locals>.select at 0x...>
>>> parse_query('limit=10')
Traceback (most recent call last):
...
starlette.exceptions.HTTPException: 400
# >>> parse_query('limit=10')
# starlette.exceptions.HTTPException: 400
"""

View File

@ -23,7 +23,9 @@ from io import BytesIO
import orjson
# asgi framework
from starlette.responses import PlainTextResponse, Response, JSONResponse
from starlette.responses import PlainTextResponse, Response, JSONResponse, HTMLResponse
from starlette.requests import Request
from starlette.exceptions import HTTPException
from .user import JWTUser, Nobody
from ..logging import logger
@ -118,6 +120,8 @@ class HJSONResponse(ORJSONResponse):
return super().render(list(content))
class ODSResponse(Response):
file_type = 'ods'
def __init__(self, d_rows: typing.List[typing.Dict]):
try:
import pyexcel as pe
@ -136,14 +140,14 @@ class ODSResponse(Response):
for elt in d_rows:
rows.append(list(elt.values()))
rows.insert(rows_names)
rows.insert(0, rows_names)
self.sheet = pe.Sheet(rows)
self.sheet.save_to_memory(
file_type='ods',
file_type=self.file_type,
stream=ods_file)
filename = f'{date.today()}.ods'
filename = f'{date.today()}.{self.file_type}'
super().__init__(
content=ods_file.getvalue(),
@ -152,3 +156,12 @@ class ODSResponse(Response):
'Content-Disposition': f'attachment; filename="{filename}"'},
status_code = 200)
class XLSXResponse(ODSResponse):
file_type = 'xlsx'
def gen_exception_route(response_cls):
async def exception_route(req: Request, exc: HTTPException):
return response_cls()
return exception_route

View File

@ -2,11 +2,13 @@
"""
Routes module
Classes :
- JSONRoute
Fonctions :
- gen_domain_routes
- gen_starlette_routes
- gen_schema_routes
- api_routes
- api_acls
Exception :
- DomainNotFoundError
@ -43,7 +45,6 @@ def JSONRoute(data: Any) -> Coroutine:
async function
"""
async def wrapped(request, *args, **kwargs):
logger.debug('JSONRoute data: %s', data)
return ORJSONResponse(data)
return wrapped
@ -84,20 +85,6 @@ def gen_schema_routes(schema: Dict):
yield HalfRoute(path, args_check(fct), acls, verb)
def gen_starlette_routes(d_domains: Dict[str, ModuleType]) -> Generator:
"""
Yields the Route objects for HalfAPI app
Parameters:
d_domains (dict[str, ModuleType])
Returns:
Generator(Route)
"""
for domain_name, m_domain in d_domains.items():
yield from gen_domain_routes(m_domain)
def api_routes(m_dom: ModuleType) -> Tuple[Dict, Dict]:
"""
Yields the description objects for HalfAPI app routes
@ -147,28 +134,3 @@ def api_routes(m_dom: ModuleType) -> Tuple[Dict, Dict]:
raise exc
return d_res, d_acls
def api_acls(request):
""" Returns the list of possible ACLs
# TODO: Rewrite
"""
res = {}
domains = {}
doc = 'doc' in request.query_params
for domain, m_domain in domains.items():
res[domain] = {}
for acl_name, fct in domain_acls(m_domain, [domain]):
if not isinstance(fct, FunctionType):
continue
fct_result = fct.__doc__.strip() if doc and fct.__doc__ else fct(request)
if acl_name in res[domain]:
continue
if isinstance(fct_result, FunctionType):
fct_result = fct()(request)
res[domain][acl_name] = fct_result
return res

View File

@ -13,13 +13,13 @@ import os
import importlib
from typing import Dict, Coroutine, List
from types import ModuleType
import yaml
from starlette.schemas import SchemaGenerator
from .. import __version__
# from .domain import gen_router_routes, domain_schema_list
from ..logging import logger
from .routes import gen_starlette_routes, api_routes, api_acls
from .routes import api_routes
from .responses import ORJSONResponse
SCHEMAS = SchemaGenerator(
@ -28,74 +28,17 @@ SCHEMAS = SchemaGenerator(
async def schema_json(request, *args, **kwargs):
"""
description: Returns the current API routes description (OpenAPI v3)
as a JSON object
description: |
Returns the current API routes description (OpenAPI v3)
as a JSON object
responses:
200:
description: API Schema in OpenAPI v3 format
"""
return ORJSONResponse(
SCHEMAS.get_schema(routes=request.app.routes))
def schema_dict_dom(d_domains: Dict[str, ModuleType]) -> Dict:
"""
Returns the API schema of the *m_domain* domain as a python dictionnary
Parameters:
d_domains (Dict[str, moduleType]): The module to scan for routes
Returns:
Dict: A dictionnary containing the description of the API using the
| OpenAPI standard
"""
return SCHEMAS.get_schema(
routes=list(gen_starlette_routes(d_domains)))
async def get_acls(request, *args, **kwargs):
"""
description: A dictionnary of the domains and their acls, with the
result of the acls functions
"""
return ORJSONResponse(api_acls(request))
def schema_to_csv(module_name, header=True) -> str:
"""
Returns a string composed where each line is a set of path, verb, function,
acl, required arguments, optional arguments and output variables. Those
lines should be unique in the result string;
"""
# retrieve module
m_router = importlib.import_module(module_name)
lines = []
if header:
lines.append([
'path',
'method',
'module:function',
'acl',
'args_required', 'args_optional',
'out'
])
for line in domain_schema_list(m_router):
lines.append([
line[0],
line[1],
line[2],
line[3],
','.join(line[4]),
','.join(line[5]),
','.join(line[6])
])
return '\n'.join(
[ ';'.join(fields) for fields in lines ]
)
def schema_csv_dict(csv: List[str], prefix='/') -> Dict:
package = None
schema_d = {}
@ -172,3 +115,23 @@ def schema_csv_dict(csv: List[str], prefix='/') -> Dict:
})
return schema_d
def param_docstring_default(name, type):
""" Returns a default docstring in OpenAPI format for a path parameter
"""
type_map = {
'str': 'string',
'uuid': 'string',
'path': 'string',
'int': 'number',
'float': 'number'
}
return yaml.dump({
'name': name,
'in': 'path',
'description': f'default description for path parameter {name}',
'required': True,
'schema': {
'type': type_map[type]
}
})

View File

@ -1,8 +1,10 @@
import logging
default_level = logging.DEBUG
default_format = '%(asctime)s [%(process)d] [%(levelname)s] %(message)s'
default_datefmt = '[%Y-%m-%d %H:%M:%S %z]'
def config_logging(level=logging.INFO):
def config_logging(level=default_level, format=default_format, datefmt=default_datefmt):
# When run by 'uvicorn ...', a root handler is already
# configured and the basicConfig below does nothing.
# To get the desired formatting:
@ -12,8 +14,8 @@ def config_logging(level=logging.INFO):
# https://github.com/encode/uvicorn/issues/511
logging.basicConfig(
# match gunicorn format
format='%(asctime)s [%(process)d] [%(levelname)s] %(message)s',
datefmt='[%Y-%m-%d %H:%M:%S %z]',
format=format,
datefmt=datefmt,
level=level)
# When run by 'gunicorn -k uvicorn.workers.UvicornWorker ...',
@ -27,5 +29,4 @@ def config_logging(level=logging.INFO):
logging.getLogger('uvicorn.access').propagate = True
logging.getLogger('uvicorn.error').propagate = True
config_logging()
logger = logging.getLogger()

View File

@ -3,18 +3,39 @@ import functools
import os
import sys
import json
from json.decoder import JSONDecodeError
import toml
from unittest import TestCase
from starlette.testclient import TestClient
from click.testing import CliRunner
from ..cli.cli import cli
from ..halfapi import HalfAPI
from ..half_domain import HalfDomain
from ..conf import DEFAULT_CONF
from pprint import pprint
import tempfile
class TestDomain(TestCase):
@property
def domain_name(self):
return getattr(self, 'DOMAIN')
@property
def module_name(self):
return getattr(self, 'MODULE', self.domain_name)
@property
def acl_path(self):
return getattr(self, 'ACL', '.acl')
@property
def router_path(self):
return getattr(self, 'ROUTERS', '.routers')
@property
def router_module(self):
return '.'.join((self.DOMAIN, self.ROUTERS))
return '.'.join((self.module_name, self.ROUTERS))
def setUp(self):
# CLI
@ -39,27 +60,37 @@ class TestDomain(TestCase):
return wrapper
class_.invoke = invoke_wrapper(class_.invoke)
self.runner = class_(mix_stderr=False)
self.runner = class_()
# HTTP
# Fake default values of default configuration
self.halfapi_conf = {
'secret': 'testsecret',
'production': False,
'domain': {}
}
self.halfapi_conf['domain'][self.DOMAIN] = {
'name': self.DOMAIN,
'router': self.ROUTERS,
self.halfapi_conf['domain'][self.domain_name] = {
'name': self.domain_name,
'router': self.router_path,
'acl': self.acl_path,
'module': self.module_name,
'prefix': False,
'enabled': True,
'config': {
'test': True
}
'config': getattr(self, 'CONFIG', {})
}
_, self.config_file = tempfile.mkstemp()
with open(self.config_file, 'w') as fh:
fh.write(toml.dumps(self.halfapi_conf))
self.halfapi = HalfAPI(self.halfapi_conf)
self.client = TestClient(self.halfapi.application)
self.module = importlib.import_module(
self.module_name
)
def tearDown(self):
@ -70,43 +101,52 @@ class TestDomain(TestCase):
try:
result = self.runner.invoke(cli, '--version')
self.assertEqual(result.exit_code, 0)
result = self.runner.invoke(cli, ['domain', self.DOMAIN])
result = self.runner.invoke(cli, ['domain', '--read', self.DOMAIN, self.config_file])
self.assertEqual(result.exit_code, 0)
result_d = json.loads(result.stdout)
result = self.runner.invoke(cli, ['run', '--help'])
self.assertEqual(result.exit_code, 0)
result = self.runner.invoke(cli, ['run', '--dryrun', self.DOMAIN])
self.assertEqual(result.exit_code, 0)
# result = self.runner.invoke(cli, ['run', '--dryrun', self.DOMAIN])
# self.assertEqual(result.exit_code, 0)
except AssertionError as exc:
print(f'Result {result}')
print(f'Stdout {result.stdout}')
print(f'Stderr {result.stderr}')
raise exc
except JSONDecodeError as exc:
print(f'Result {result}')
print(f'Stdout {result.stdout}')
raise exc
return result_d
def check_routes(self):
r = self.client.get('/')
r = self.client.request('get', '/')
assert r.status_code == 200
schemas = r.json()
assert isinstance(schemas, list)
for schema in schemas:
assert isinstance(schema, dict)
assert 'openapi' in schema
assert 'info' in schema
assert 'paths' in schema
assert 'domain' in schema
schema = r.json()
assert isinstance(schema, dict)
assert 'openapi' in schema
assert 'info' in schema
assert 'paths' in schema
r = self.client.get('/halfapi/acls')
r = self.client.request('get', '/halfapi/acls')
assert r.status_code == 200
d_r = r.json()
assert isinstance(d_r, dict)
assert self.domain_name in d_r.keys()
assert self.DOMAIN in d_r.keys()
ACLS = HalfDomain.acls(self.module, self.acl_path)
assert len(ACLS) == len(d_r[self.domain_name])
ACLS = HalfDomain.acls(self.DOMAIN)
assert len(ACLS) == len(d_r[self.DOMAIN])
for acl_rule in ACLS:
assert len(acl_rule.name) > 0
assert acl_rule.name in d_r[self.domain_name]
assert len(acl_rule.documentation) > 0
assert isinstance(acl_rule.priority, int)
assert acl_rule.priority >= 0
for acl_name in ACLS:
assert acl_name[0] in d_r[self.DOMAIN]
if acl_rule.public is True:
r = self.client.request('get', f'/halfapi/acls/{acl_rule.name}')
assert r.status_code in [200, 401]

View File

@ -1,2 +1,3 @@
pyexcel>=0.6.3,<1
pyexcel-ods>=0.5.6,<1
pyexcel-xlsx=0.6.0,<1

View File

@ -1,5 +1,5 @@
[pytest]
testpaths = tests
testpaths = tests halfapi
addopts = --doctest-modules
doctest_optionflags = ELLIPSIS
python_paths = ./tests
doctest_optionflags = ELLIPSIS IGNORE_EXCEPTION_DETAIL
pythonpath = ./tests

View File

@ -1,5 +1,61 @@
click
starlette
uvicorn
PyJWT
pygit2==0.28.2
alog==0.9.13
anyio==3.4.0
asgiref==3.4.1
astroid==2.9.0
attrs==21.2.0
bleach==4.1.0
build==0.7.0
certifi==2021.10.8
cffi==1.15.0
charset-normalizer==2.0.9
click==7.1.2
colorama==0.4.4
contextlib2==21.6.0
cryptography==36.0.1
docutils==0.18.1
h11==0.12.0
idna==3.3
importlib-metadata==4.8.2
iniconfig==1.1.1
isort==5.10.1
jeepney==0.7.1
keyring==23.4.0
lazy-object-proxy==1.7.1
mccabe==0.6.1
orjson==3.6.5
packaging==21.3
pep517==0.12.0
pkginfo==1.8.2
platformdirs==2.4.0
pluggy==1.0.0
py==1.11.0
pycparser==2.21
pyflakes==2.4.0
Pygments==2.10.0
PyJWT==2.3.0
pylint==2.12.2
pyparsing==3.0.6
pytest==6.2.5
pytest-asyncio==0.16.0
pytest-pythonpath==0.7.3
PyYAML==5.4.1
readme-renderer==32.0
requests==2.26.0
requests-toolbelt==0.9.1
rfc3986==1.5.0
schema==0.7.5
SecretStorage==3.3.1
six==1.16.0
sniffio==1.2.0
starlette==0.17.1
timing-asgi==0.2.1
toml==0.10.2
tomli==2.0.0
tqdm==4.62.3
twine==3.7.1
urllib3==1.26.7
uvicorn==0.16.0
vulture==2.3
webencodings==0.5.1
wrapt==1.13.3
zipp==3.6.0

View File

@ -43,30 +43,43 @@ setup(
packages=get_packages(module_name),
python_requires=">=3.8",
install_requires=[
"PyJWT>=2.3.0,<2.4.0",
"starlette>=0.17,<0.18",
"click>=7.1,<8",
"PyJWT>=2.6.0,<2.7.0",
"starlette>=0.33,<0.34",
"click>=8,<9",
"uvicorn>=0.13,<1",
"orjson>=3.4.7,<4",
"pyyaml>=5.3.1,<6",
"orjson>=3.8.5,<4",
"pyyaml>=6,<7",
"timing-asgi>=0.2.1,<1",
"schema>=0.7.4,<1",
"toml>=0.7.1,<0.8"
"toml>=0.10,<0.11",
"packaging>=19.0",
"python-multipart"
],
classifiers=[
"Development Status :: 3 - Alpha",
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
"Intended Audience :: Developers",
"Topic :: Software Development :: Libraries :: Application Frameworks",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10"
],
extras_require={
"tests":[
"pytest",
"requests",
"pytest>=7,<8",
"pytest-asyncio",
"pylint"
"pylint",
"requests",
"httpx",
"openapi-schema-validator",
"openapi-spec-validator",
"coverage"
],
"pyexcel":[
"pyexcel",
"pyexcel-ods3",
"pyexcel-xlsx"
]
},
entry_points={

View File

@ -4,10 +4,13 @@ import subprocess
import importlib
import tempfile
from unittest.mock import patch
import json
import toml
import pytest
from click.testing import CliRunner
from configparser import ConfigParser
from halfapi.conf import DEFAULT_CONF, PROJECT_LEVEL_KEYS, DOMAIN_LEVEL_KEYS
PROJNAME = os.environ.get('PROJ','tmp_api')
@ -22,10 +25,51 @@ class TestCliProj():
def test_domain_commands(self, project_runner):
""" TODO: Test create command
"""
test_conf = {
'project': {
'port': '3010',
'loglevel': 'warning'
},
'domain': {
'dummy_domain': {
'port': 4242,
'name': 'dummy_domain',
'enabled': True
}
}
}
r = project_runner('domain')
print(r.stdout)
assert r.exit_code == 1
r = project_runner('domain dummy_domain')
_, tmp_conf = tempfile.mkstemp()
with open(tmp_conf, 'w') as fh:
fh.write(
toml.dumps(test_conf)
)
r = project_runner(f'domain dummy_domain --conftest {tmp_conf}')
assert r.exit_code == 0
r_conf = toml.loads(r.stdout)
for key, value in r_conf.items():
if key == 'domain':
continue
assert key in PROJECT_LEVEL_KEYS
if key == 'port':
assert value == test_conf['domain']['dummy_domain']['port']
elif key == 'loglevel':
assert value == test_conf['project']['loglevel']
else:
assert value == DEFAULT_CONF[key.upper()]
assert json.dumps(test_conf['domain']) == json.dumps(r_conf['domain'])
for key in test_conf['domain']['dummy_domain'].keys():
assert key in DOMAIN_LEVEL_KEYS
# Default command "run"
r = project_runner(f'domain dummy_domain --dry-run {tmp_conf}')
print(r.stdout)
assert r.exit_code == 0

View File

@ -1,9 +1,11 @@
import pytest
from click.testing import CliRunner
from halfapi.cli.cli import cli
import os
from unittest.mock import patch
@pytest.mark.skip
def test_run_noproject(cli_runner):
with cli_runner.isolated_filesystem():
result = cli_runner.invoke(cli, ['config'])

View File

@ -76,7 +76,7 @@ def token_debug_true_builder():
@pytest.fixture
def runner():
return CliRunner(mix_stderr=False)
return CliRunner()
@pytest.fixture
@ -104,7 +104,7 @@ def cli_runner():
return wrapper
class_.invoke = invoke_wrapper(class_.invoke)
cli_runner_ = class_(mix_stderr=False)
cli_runner_ = class_()
yield cli_runner_

View File

@ -1,6 +1,27 @@
__name__ = 'dummy_domain'
__version__ = '0.0.0'
__patch_release__ = '0.0.0'
__routers__ = '.routers'
__id__ = '8b88e60a625369235b36c2d6d70756a0c02c1c7fb169fcee6dc820bcf9723f5a'
from halfapi import __version__ as halfapi_version
domain = {
'name': 'dummy_domain',
'version': '0.0.0',
'id': '8b88e60a625369235b36c2d6d70756a0c02c1c7fb169fcee6dc820bcf9723f5a',
'deps': (
('halfapi', '=={}'.format(halfapi_version)),
),
'schema_components': {
'schemas': {
'Pinnochio': {
'type': 'object',
'required': {
'id',
'name',
'nose_size'
},
'properties': {
'id': {'type': 'string'},
'name': {'type': 'string'},
'nose_size': {'type': 'number'}
}
}
}
}
}

View File

@ -1,10 +0,0 @@
from halfapi.lib import acl
ACLS = {
'GET' : [{'acl':acl.public}]
}
def get():
"""
description:
Not implemented
"""
raise NotImplementedError

View File

@ -1,5 +1,5 @@
from halfapi.lib import acl
from halfapi.lib.acl import public, private
from halfapi.lib.acl import public, private, ACLS
from random import randint
def random(*args):
@ -7,8 +7,7 @@ def random(*args):
"""
return randint(0,1) == 1
PRIORITY = (
[public],
[random1, random2],
[private]
ACLS = (
*ACLS,
('random', random.__doc__, 10)
)

View File

@ -1,17 +0,0 @@
from ... import acl
from halfapi.logging import logger
ACLS = {
'GET' : [
{'acl':acl.public},
{'acl':acl.random},
]
}
def get(halfapi):
"""
description:
returns the configuration of the domain
"""
logger.error('%s', halfapi)
return halfapi['config']['domain']['dummy_domain']['config']

View File

@ -12,6 +12,9 @@ async def get(test):
"""
description:
returns the path parameter
responses:
200:
description: test response
"""
return ORJSONResponse(str(test))
@ -19,6 +22,9 @@ def post(test):
"""
description:
returns the path parameter
responses:
200:
description: test response
"""
return str(test)
@ -26,6 +32,9 @@ def patch(test):
"""
description:
returns the path parameter
responses:
200:
description: test response
"""
return str(test)
@ -33,6 +42,9 @@ def put(test):
"""
description:
returns the path parameter
responses:
200:
description: test response
"""
return str(test)
@ -40,5 +52,8 @@ def delete(test):
"""
description:
returns the path parameter
responses:
200:
description: test response
"""
return str(test)

View File

@ -0,0 +1,21 @@
from uuid import uuid4
from halfapi.lib import acl
ACLS = {
'GET' : [{'acl':acl.public}]
}
def get():
"""
description: The pinnochio guy
responses:
200:
description: test response
content:
application/json:
schema:
$ref: "#/components/schemas/Pinnochio"
"""
return {
'id': str(uuid4()),
'name': 'pinnochio',
'nose_size': 42
}

View File

@ -47,7 +47,8 @@ ACLS = {
'foo', 'baz'
},
'optional': {
'truebidoo'
'truebidoo',
'z'
}
}
},
@ -55,18 +56,24 @@ ACLS = {
}
def get(halfapi, data):
def get(data):
"""
description:
returns the configuration of the domain
returns the arguments passed in
responses:
200:
description: test response
"""
logger.error('%s', data['foo'])
return {'foo': data['foo'], 'bar': data['bar']}
return data
def post(halfapi, data):
def post(data):
"""
description:
returns the configuration of the domain
returns the arguments passed in
responses:
200:
description: test response
"""
logger.error('%s', data)
return {'foo': data['foo'], 'bar': data.get('bar', data.get('baz'))}
return data

View File

@ -25,24 +25,43 @@ ROUTES = {
async def get_abc_alphabet_TEST(request, *args, **kwargs):
"""
description: Not implemented
responses:
200:
description: test response
parameters:
- name: test
in: path
description: Test parameter in route with "ROUTES" constant
required: true
schema:
type: string
"""
return NotImplementedResponse()
async def get_abc_pinnochio(request, *args, **kwargs):
"""
description: Not implemented
responses:
200:
description: test response
"""
return NotImplementedResponse()
async def get_config(request, *args, **kwargs):
"""
description: Not implemented
responses:
200:
description: test response
"""
return NotImplementedResponse()
async def get_arguments(request, *args, **kwargs):
"""
description: Liste des datatypes.
responses:
200:
description: test response
"""
return ORJSONResponse({
'foo': kwargs.get('data').get('foo'),

View File

@ -0,0 +1,33 @@
from ... import acl
from halfapi.logging import logger
ACLS = {
'GET' : [
{'acl':acl.public},
{'acl':acl.random},
]
}
def get(halfapi):
"""
description:
returns the configuration of the domain
responses:
200:
description: test response
"""
logger.error('%s', halfapi)
# TODO: Remove in 0.7.0
try:
assert 'test' in halfapi['config']['domain']['dummy_domain']['config']
except AssertionError as exc:
logger.error('No TEST in halfapi[config][domain][dummy_domain][config]')
raise exc
try:
assert 'test' in halfapi['config']
except AssertionError as exc:
logger.error('No TEST in halfapi[config]')
raise exc
return halfapi['config']

View File

@ -0,0 +1,8 @@
param_docstring = """
name: second
in: path
description: second parameter description test
required: true
schema:
type: string
"""

View File

@ -0,0 +1,20 @@
from uuid import UUID
from halfapi.lib import acl
ACLS = {
'GET': [{'acl': acl.public}]
}
def get(first, second, third):
"""
description: a Test route for path parameters
responses:
200:
description: The test passed!
500:
description: The test did not pass :(
"""
assert isintance(first, str)
assert isintance(second, UUID)
assert isintance(third, int)
return ''

View File

@ -0,0 +1,13 @@
from halfapi.lib import acl
ACLS = {
'GET': [{'acl':acl.public}]
}
def get(ext, ret_type='html'):
"""
responses:
200:
description: dummy abc.alphabet route
"""
return '\n'.join(('trololo', '', 'ololotr'))

View File

@ -0,0 +1,31 @@
from halfapi.lib import acl
ACLS = {
'GET': [{'acl':acl.public}],
'POST': [
{
'acl':acl.public,
'args': {
'required': {'trou'},
'optional': {'troo'}
}
}
]
}
def get(ext, halfapi={}, ret_type='html'):
"""
responses:
200:
description: dummy abc.alphabet route
"""
return '\n'.join(('trololo', '', 'ololotr'))
def post(ext, data={'troo': 'fidget'}, halfapi={}, ret_type='html'):
"""
responses:
200:
description: dummy abc.alphabet route
"""
print(data)
return '\n'.join(('trololo', '', 'ololotr'))

View File

@ -0,0 +1,13 @@
from halfapi.lib import acl
ACLS = {
'GET': [{'acl':acl.public}]
}
def get(ret_type='html'):
"""
responses:
200:
description: dummy abc.alphabet route
"""
return '\n'.join(('trololo', '', 'ololotr'))

View File

@ -0,0 +1,4 @@
from . import get
def test_get():
assert isinstance(get(), str)

View File

@ -17,7 +17,7 @@ def test_acl_Check(dummy_app, token_debug_false_builder):
dummy_app.add_route('/test_public', test_route_public)
test_client = TestClient(dummy_app)
resp = test_client.get('/test_public?check')
resp = test_client.request('get', '/test_public?check')
assert resp.status_code == 200
@HalfRoute.acl_decorator(params=[{'acl':acl.private}])
@ -28,10 +28,10 @@ def test_acl_Check(dummy_app, token_debug_false_builder):
dummy_app.add_route('/test_private', test_route_private)
test_client = TestClient(dummy_app)
resp = test_client.get('/test_private')
resp = test_client.request('get', '/test_private')
assert resp.status_code == 401
resp = test_client.get('/test_private?check')
resp = test_client.request('get', '/test_private?check')
assert resp.status_code == 200

View File

@ -7,6 +7,10 @@ import json
import os
import sys
import pprint
import openapi_spec_validator
import logging
logger = logging.getLogger()
from halfapi.lib.constants import API_SCHEMA
@ -15,7 +19,7 @@ def test_halfapi_whoami(application_debug):
# So we use a single function with fixture "application debug"
c = TestClient(application_debug)
r = c.get('/halfapi/whoami')
r = c.request('get', '/halfapi/whoami')
assert r.status_code == 200
def test_halfapi_log(application_debug):
@ -24,7 +28,7 @@ def test_halfapi_log(application_debug):
c = TestClient(application_debug)
r = c.get('/halfapi/log')
r = c.request('get', '/halfapi/log')
assert r.status_code == 200
def test_halfapi_error_400(application_debug):
@ -33,7 +37,7 @@ def test_halfapi_error_400(application_debug):
c = TestClient(application_debug)
r = c.get('/halfapi/error/400')
r = c.request('get', '/halfapi/error/400')
assert r.status_code == 400
def test_halfapi_error_404(application_debug):
@ -42,7 +46,7 @@ def test_halfapi_error_404(application_debug):
c = TestClient(application_debug)
r = c.get('/halfapi/error/404')
r = c.request('get', '/halfapi/error/404')
assert r.status_code == 404
def test_halfapi_error_500(application_debug):
@ -51,15 +55,13 @@ def test_halfapi_error_500(application_debug):
c = TestClient(application_debug)
r = c.get('/halfapi/error/500')
r = c.request('get', '/halfapi/error/500')
assert r.status_code == 500
def test_schema(application_debug):
c = TestClient(application_debug)
r = c.get('/')
schemas = r.json()
assert isinstance(schemas, list)
for schema in schemas:
assert isinstance(schema, dict)
assert API_SCHEMA.validate(schema)
r = c.request('get', '/')
schema = r.json()
assert isinstance(schema, dict)
openapi_spec_validator.validate_spec(schema)

View File

@ -1,13 +1,106 @@
import pytest
from halfapi.testing.test_domain import TestDomain
from dummy_domain import __name__, __routers__
from pprint import pprint
import logging
logger = logging.getLogger()
class TestDummyDomain(TestDomain):
from .dummy_domain import domain
__name__ = domain.get('name')
__routers__ = domain.get('routers')
DOMAIN = __name__
ROUTERS = __routers__
CONFIG = {'test': True}
def test_domain(self):
self.check_domain()
def test_routes(self):
self.check_routes()
def test_html_route(self):
res = self.client.request('get', '/ret_type')
assert res.status_code == 200
assert isinstance(res.content.decode(), str)
assert res.headers['content-type'].split(';')[0] == 'text/html'
res = self.client.request('get', '/ret_type/h24')
assert res.status_code == 200
assert isinstance(res.content.decode(), str)
assert res.headers['content-type'].split(';')[0] == 'text/html'
res = self.client.request('get', '/ret_type/h24/config')
assert res.status_code == 200
assert isinstance(res.content.decode(), str)
assert res.headers['content-type'].split(';')[0] == 'text/html'
res = self.client.request('post', '/ret_type/h24/config', json={
'trou': 'glet'
})
assert res.status_code == 200
assert isinstance(res.content.decode(), str)
assert res.headers['content-type'].split(';')[0] == 'text/html'
def test_arguments__get_routes(self):
res = self.client.request('post', '/arguments?foo=1&x=3')
assert res.status_code == 400
arg_dict = {'foo': '1', 'bar': '2', 'x': '3'}
res = self.client.request('get', '/arguments?foo=1&bar=2&x=3')
assert res.json() == arg_dict
res = self.client.request('get', '/arguments?foo=1&bar=2&x=3&y=4')
assert res.json() == arg_dict
def test_arguments_post_routes(self):
arg_dict = {}
res = self.client.request('post', '/arguments', json=arg_dict)
assert res.status_code == 400
arg_dict = {'foo': '1', 'bar': '3'}
res = self.client.request('post', '/arguments', json=arg_dict)
assert res.status_code == 400
arg_dict = {'foo': '1', 'baz': '3'}
res = self.client.request('post', '/arguments', json=arg_dict)
assert res.json() == arg_dict
arg_dict = {'foo': '1', 'baz': '3', 'truebidoo': '4'}
res = self.client.request('post', '/arguments', json=arg_dict)
assert res.json() == arg_dict
res = self.client.request('post', '/arguments', json={ **arg_dict, 'y': '4'})
assert res.json() == arg_dict
res = self.client.request('post', '/arguments', json={ **arg_dict, 'z': True})
assert res.json() == {**arg_dict, 'z': True}
def test_schema_path_params(self):
res = self.client.request('get', '/halfapi/schema')
schema = res.json()
logger.debug(schema)
assert len(schema['paths']) > 0
route = schema['paths']['/path_params/{first}/one/{second}/two/{third}']
assert 'parameters' in route['get']
parameters = route['get']['parameters']
assert len(parameters) == 3
param_map = {
elt['name']: elt
for elt in parameters
}
assert param_map['second']['description'] == 'second parameter description test'

View File

@ -1,4 +1,5 @@
import importlib
from halfapi.testing.test_domain import TestDomain
def test_dummy_domain():
from . import dummy_domain
@ -13,7 +14,6 @@ def test_dummy_domain():
from .dummy_domain.routers.abc.alphabet.TEST_uuid import get
from .dummy_domain.routers.abc.pinnochio import get
from .dummy_domain.routers.config import get
async_mod = importlib.import_module('dummy_domain.routers.async', '.')
fcts = ['get_abc_alphabet_TEST', 'get_abc_pinnochio', 'get_config', 'get_arguments']
for fct in fcts:
getattr(async_mod, fct)
from .dummy_domain.routers.config import get
from .dummy_domain.routers import async_router
from .dummy_domain.routers.async_router import ROUTES, get_abc_alphabet_TEST, get_abc_pinnochio, get_config, get_arguments

View File

@ -4,16 +4,17 @@ import importlib
import subprocess
import time
import pytest
import json
from pprint import pprint
from starlette.routing import Route
from starlette.testclient import TestClient
def test_get_config_route(dummy_project, application_domain):
c = TestClient(application_domain)
r = c.get('/')
r = c.request('get', '/')
assert r.status_code == 200
pprint(r.json())
r = c.get('/config')
r = c.request('get', '/config')
assert r.status_code == 200
pprint(r.json())
assert 'test' in r.json()
@ -24,8 +25,8 @@ def test_get_route(dummy_project, application_domain):
dummy_domain_routes = [
('config','GET'),
('config','GET'),
('async/abc/pinnochio','GET'),
('async/config','GET'),
('async_router/abc/pinnochio','GET'),
('async_router/config','GET'),
# ('abc/pinnochio','GET'),
# ('abc/alphabet','GET'),
]
@ -36,15 +37,15 @@ def test_get_route(dummy_project, application_domain):
print(route_path)
try:
if verb.lower() == 'get':
r = c.get(route_path)
r = c.request('get', route_path)
elif verb.lower() == 'post':
r = c.post(route_path)
r = c.request('post', route_path)
elif verb.lower() == 'patch':
r = c.patch(route_path)
r = c.request('patch', route_path)
elif verb.lower() == 'put':
r = c.put(route_path)
r = c.request('put', route_path)
elif verb.lower() == 'delete':
r = c.delete(route_path)
r = c.request('delete', route_path)
else:
raise Exception(verb)
try:
@ -69,7 +70,7 @@ def test_get_route(dummy_project, application_domain):
path = path.format(test=str(test_uuid))
route_path = f'/{path}'
if verb.lower() == 'get':
r = c.get(f'{route_path}')
r = c.request('get', f'{route_path}')
assert r.status_code == 200
@ -78,7 +79,7 @@ def test_delete_route(dummy_project, application_domain):
c = TestClient(application_domain)
from uuid import uuid4
arg = str(uuid4())
r = c.delete(f'/abc/alphabet/{arg}')
r = c.request('delete', f'/abc/alphabet/{arg}')
assert r.status_code == 200
assert isinstance(r.json(), str)
@ -86,23 +87,23 @@ def test_arguments_route(dummy_project, application_domain):
c = TestClient(application_domain)
path = '/arguments'
r = c.get(path)
r = c.request('get', path)
assert r.status_code == 400
r = c.get(path, params={'foo':True})
r = c.request('get', path, params={'foo':True})
assert r.status_code == 400
arg = {'foo':True, 'bar':True}
r = c.get(path, params=arg)
r = c.request('get', path, params=arg)
assert r.status_code == 200
for key, val in arg.items():
assert r.json()[key] == str(val)
path = '/async/arguments'
r = c.get(path)
assert json.loads(r.json()[key]) == val
path = '/async_router/arguments'
r = c.request('get', path)
assert r.status_code == 400
r = c.get(path, params={'foo':True})
r = c.request('get', path, params={'foo':True})
assert r.status_code == 400
arg = {'foo':True, 'bar':True}
r = c.get(path, params=arg)
r = c.request('get', path, params=arg)
assert r.status_code == 200
for key, val in arg.items():
assert r.json()[key] == str(val)
assert json.loads(r.json()[key]) == val

View File

@ -38,7 +38,7 @@ def test_jwt_NoToken(dummy_app):
dummy_app.add_route('/test', test_route)
test_client = TestClient(dummy_app)
resp = test_client.get('/test')
resp = test_client.request('get', '/test')
assert resp.status_code == 200
def test_jwt_Token(dummy_app, token_builder):
@ -50,13 +50,20 @@ def test_jwt_Token(dummy_app, token_builder):
dummy_app.add_route('/test', test_route)
test_client = TestClient(dummy_app)
resp = test_client.get('/test',
resp = test_client.request('get', '/test',
cookies={
'Authorization': token_builder
})
assert resp.status_code == 200
resp = test_client.request('get', '/test',
headers={
'Authorization': token_builder
})
assert resp.status_code == 200
def test_jwt_DebugFalse(dummy_app, token_debug_false_builder):
async def test_route(request):
assert isinstance(request.user, JWTUser)
@ -65,7 +72,13 @@ def test_jwt_DebugFalse(dummy_app, token_debug_false_builder):
dummy_app.add_route('/test', test_route)
test_client = TestClient(dummy_app)
resp = test_client.get('/test',
resp = test_client.request('get', '/test',
cookies={
'Authorization': token_debug_false_builder
})
assert resp.status_code == 200
resp = test_client.request('get', '/test',
headers={
'Authorization': token_debug_false_builder
})
@ -82,7 +95,13 @@ def test_jwt_DebugTrue(dummy_app, token_debug_true_builder):
dummy_app.add_route('/test', test_route)
test_client = TestClient(dummy_app)
resp = test_client.get('/test',
resp = test_client.request('get', '/test',
cookies={
'Authorization': token_debug_true_builder
})
assert resp.status_code == 400
resp = test_client.request('get', '/test',
headers={
'Authorization': token_debug_true_builder
})
@ -100,8 +119,14 @@ def test_jwt_DebugTrue_DebugApp(dummy_debug_app, token_debug_true_builder):
dummy_debug_app.add_route('/test', test_route)
test_client = TestClient(dummy_debug_app)
resp = test_client.get('/test',
headers={
'Authorization': token_debug_true_builder
resp = test_client.request('get', '/test',
cookies={
'Authorization': token_debug_true_builder
})
assert resp.status_code == 200
resp = test_client.request('get', '/test',
headers={
'Authorization': token_debug_true_builder
})
assert resp.status_code == 200

View File

@ -49,3 +49,76 @@
# 
#  assert isinstance(res, list)
#  assert len(res) > 0
from starlette.testclient import TestClient
from starlette.responses import Response
from starlette.routing import Router, Route
from halfapi.lib.domain import route_decorator
from halfapi.lib.user import Nobody
def test_route_decorator():
""" It should decorate an async function that fullfills its arguments
"""
def route(halfapi, data, out, ret_type='txt'):
for key in ['user', 'config', 'domain', 'cookies', 'base_url', 'url']:
assert key in halfapi
assert halfapi['user'] is None
assert isinstance(halfapi['config'], dict)
assert len(halfapi['config']) == 0
assert isinstance(halfapi['domain'], str)
assert halfapi['domain'] == 'unknown'
assert isinstance(halfapi['cookies'], dict)
assert len(halfapi['cookies']) == 0
assert len(str(halfapi['base_url'])) > 0
assert str(halfapi['base_url']) == 'http://testserver/'
assert len(str(halfapi['url'])) > 0
assert str(halfapi['url']) == 'http://testserver/'
assert isinstance(data, dict)
assert len(data) == 0
assert out is None
assert ret_type is 'txt'
return ''
async_route = route_decorator(route)
app = Router([Route('/', endpoint=async_route, methods=['GET'])])
client = TestClient(app)
response = client.get('/')
assert response.is_success
assert response.content.decode() == ''
def route(data, out, ret_type='txt'):
assert isinstance(data, dict)
assert len(data) == 0
assert out is None
assert ret_type is 'txt'
return ''
async_route = route_decorator(route)
app = Router([Route('/', endpoint=async_route, methods=['GET'])])
client = TestClient(app)
response = client.get('/')
assert response.is_success
assert response.content.decode() == ''
def route(data):
assert isinstance(data, dict)
assert len(data) == 2
assert data['toto'] == 'tata'
assert data['bouboul'] == True
return ''
async_route = route_decorator(route)
app = Router([Route('/', endpoint=async_route, methods=['POST'])])
client = TestClient(app)
response = client.post('/', json={'toto': 'tata', 'bouboul': True})
assert response.is_success
assert response.json() == ''

View File

@ -12,26 +12,27 @@ def test_init():
def test_call(application_debug):
c = TestClient(application_debug)
r = c.get('/abc/alphabet')
r = c.request('get', '/abc/alphabet')
assert r.status_code == 200
assert r.headers['x-domain'] == 'dummy_domain'
assert r.headers['x-acl'] == 'public'
r = c.get('/arguments')
r = c.request('get', '/arguments')
assert r.status_code == 400
assert r.headers['x-domain'] == 'dummy_domain'
assert r.headers['x-acl'] == 'public'
assert 'foo' in r.headers['x-args-required']
assert 'bar' in r.headers['x-args-required']
assert 'foo' in r.headers['x-args-required'].split(',')
assert 'bar' in r.headers['x-args-required'].split(',')
assert r.headers['x-args-optional'] == 'x'
c = TestClient(application_debug)
r = c.post('/arguments')
r = c.request('post', '/arguments')
assert r.status_code == 400
assert r.headers['x-domain'] == 'dummy_domain'
assert r.headers['x-acl'] == 'public'
assert 'foo' in r.headers['x-args-required']
assert 'baz' in r.headers['x-args-required']
assert r.headers['x-args-optional'] == 'truebidoo'
assert 'foo' in r.headers['x-args-required'].split(',')
assert 'baz' in r.headers['x-args-required'].split(',')
assert 'truebidoo' in r.headers['x-args-optional'].split(',')
assert 'z' in r.headers['x-args-optional'].split(',')