Add CI/CD pipeline, adjust for integration with CleverSwarm and other python services
This commit is contained in:
Stanislav Hejny
@@ -50,3 +50,6 @@ hs_err_pid*
|
||||
|
||||
**/__pycache__
|
||||
.idea
|
||||
.venv
|
||||
.coverage
|
||||
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
# This pipeline is a placeholder that does nothing.
|
||||
# This pipeline is added because GitLab requires a pipeline to success to allow merges.
|
||||
stages:
|
||||
- nop
|
||||
|
||||
nop:
|
||||
stage: nop
|
||||
script:
|
||||
- echo "nop..."
|
||||
|
||||
@@ -1,107 +1,45 @@
|
||||
|
||||
|
||||
[](https://git.cleverthis.com/cleverthis/base/root/-/commits/master)
|
||||
[](https://git.cleverthis.com/cleverthis/base/root/-/commits/master)
|
||||
[](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/-/commits/master)
|
||||
[](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/-/commits/master)
|
||||
[](https://semver.org/spec/v2.0.0.html)
|
||||
[](https://matrix.to/#/#CleverThis:qoto.org)
|
||||
|
||||
# amq-adapter-python
|
||||
|
||||
|
||||
|
||||
## Getting started
|
||||
|
||||
To make it easy for you to get started with GitLab, here's a list of recommended next steps.
|
||||
|
||||
Already a pro? Just edit this README.md and make it your own. Want to make it easy? [Use the template at the bottom](#editing-this-readme)!
|
||||
|
||||
## Add your files
|
||||
|
||||
- [ ] [Create](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#create-a-file) or [upload](https://docs.gitlab.com/ee/user/project/repository/web_editor.html#upload-a-file) files
|
||||
- [ ] [Add files using the command line](https://docs.gitlab.com/ee/gitlab-basics/add-file.html#add-a-file-using-the-command-line) or push an existing Git repository with the following command:
|
||||
|
||||
```
|
||||
cd existing_repo
|
||||
git remote add origin https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python.git
|
||||
git branch -M master
|
||||
git push -uf origin master
|
||||
```
|
||||
|
||||
## Integrate with your tools
|
||||
|
||||
- [ ] [Set up project integrations](https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python/-/settings/integrations)
|
||||
|
||||
## Collaborate with your team
|
||||
|
||||
- [ ] [Invite team members and collaborators](https://docs.gitlab.com/ee/user/project/members/)
|
||||
- [ ] [Create a new merge request](https://docs.gitlab.com/ee/user/project/merge_requests/creating_merge_requests.html)
|
||||
- [ ] [Automatically close issues from merge requests](https://docs.gitlab.com/ee/user/project/issues/managing_issues.html#closing-issues-automatically)
|
||||
- [ ] [Enable merge request approvals](https://docs.gitlab.com/ee/user/project/merge_requests/approvals/)
|
||||
- [ ] [Set auto-merge](https://docs.gitlab.com/ee/user/project/merge_requests/merge_when_pipeline_succeeds.html)
|
||||
|
||||
## Test and Deploy
|
||||
|
||||
Use the built-in continuous integration in GitLab.
|
||||
|
||||
- [ ] [Get started with GitLab CI/CD](https://docs.gitlab.com/ee/ci/quick_start/index.html)
|
||||
- [ ] [Analyze your code for known vulnerabilities with Static Application Security Testing (SAST)](https://docs.gitlab.com/ee/user/application_security/sast/)
|
||||
- [ ] [Deploy to Kubernetes, Amazon EC2, or Amazon ECS using Auto Deploy](https://docs.gitlab.com/ee/topics/autodevops/requirements.html)
|
||||
- [ ] [Use pull-based deployments for improved Kubernetes management](https://docs.gitlab.com/ee/user/clusters/agent/)
|
||||
- [ ] [Set up protected environments](https://docs.gitlab.com/ee/ci/environments/protected_environments.html)
|
||||
|
||||
***
|
||||
|
||||
# Editing this README
|
||||
|
||||
When you're ready to make this README your own, just edit this file and use the handy template below (or feel free to structure it however you want - this is just a starting point!). Thanks to [makeareadme.com](https://www.makeareadme.com/) for this template.
|
||||
|
||||
## Suggestions for a good README
|
||||
|
||||
Every project is different, so consider which of these sections apply to yours. The sections used in the template are suggestions for most open source projects. Also keep in mind that while a README can be too long and detailed, too long is better than too short. If you think your README is too long, consider utilizing another form of documentation rather than cutting out information.
|
||||
|
||||
## Name
|
||||
Choose a self-explaining name for your project.
|
||||
|
||||
## Description
|
||||
Let people know what your project can do specifically. Provide context and add a link to any reference visitors might be unfamiliar with. A list of Features or a Background subsection can also be added here. If there are alternatives to your project, this is a good place to list differentiating factors.
|
||||
|
||||
## Badges
|
||||
On some READMEs, you may see small images that convey metadata, such as whether or not all the tests are passing for the project. You can use Shields to add some to your README. Many services also have instructions for adding a badge.
|
||||
|
||||
## Visuals
|
||||
Depending on what you are making, it can be a good idea to include screenshots or even a video (you'll frequently see GIFs rather than actual videos). Tools like ttygif can help, but check out Asciinema for a more sophisticated method.
|
||||
|
||||
## Installation
|
||||
Within a particular ecosystem, there may be a common way of installing things, such as using Yarn, NuGet, or Homebrew. However, consider the possibility that whoever is reading your README is a novice and would like more guidance. Listing specific steps helps remove ambiguity and gets people to using your project as quickly as possible. If it only runs in a specific context like a particular programming language version or operating system or has dependencies that have to be installed manually, also add a Requirements subsection.
|
||||
This project contains the Python implementation of the AMQ adapter
|
||||
for routing the traffic over the RabbitMQ to intended service and automated service discovery.
|
||||
|
||||
## Usage
|
||||
Use examples liberally, and show the expected output if you can. It's helpful to have inline the smallest example of usage that you can demonstrate, while providing links to more sophisticated examples if they are too long to reasonably include in the README.
|
||||
|
||||
## Support
|
||||
Tell people where they can go to for help. It can be any combination of an issue tracker, a chat room, an email address, etc.
|
||||
### Install dependencies
|
||||
|
||||
## Roadmap
|
||||
If you have ideas for releases in the future, it is a good idea to list them in the README.
|
||||
```bash
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
## Contributing
|
||||
State if you are open to contributions and what your requirements are for accepting them.
|
||||
### Use as a library
|
||||
|
||||
For people who want to make changes to your project, it's helpful to have some documentation on how to get started. Perhaps there is a script that they should run or some environment variables that they need to set. Make these steps explicit. These instructions could also be useful to your future self.
|
||||
Use this URL to add this project as a dependency:
|
||||
|
||||
You can also document commands to lint the code or run tests. These steps help to ensure high code quality and reduce the likelihood that the changes inadvertently break something. Having instructions for running tests is especially helpful if it requires external setup, such as starting a Selenium server for testing in a browser.
|
||||
```
|
||||
git+https://git.cleverthis.com/cleverthis/clevermicro/amq-adapter-python.git@develop#egg=amq_adapter
|
||||
```
|
||||
|
||||
## Authors and acknowledgment
|
||||
Show your appreciation to those who have contributed to the project.
|
||||
> Hint: use `git config --global credential.helper store` to enable Git credential manager
|
||||
> so you don't have to re-enter the gitlab credential everytime.
|
||||
|
||||
## License
|
||||
For open source projects, say how it is licensed.
|
||||
Then:
|
||||
|
||||
## Project status
|
||||
If you have run out of energy or time for your project, put a note at the top of the README saying that development has slowed down or stopped completely. Someone may choose to fork your project or volunteer to step in as a maintainer or owner, allowing your project to keep going. You can also make an explicit request for maintainers.
|
||||
```python
|
||||
from amqp.service import amq_service
|
||||
|
||||
# TODO: how to use?
|
||||
```
|
||||
|
||||
## Donating
|
||||
|
||||
[](https://opencollective.com/cleverthis)
|
||||
|
||||
As an open-source project we run entierly off donations. Buy one of our hardworking developers a beer by [donating at OpenCollective](https://opencollective.com/cleverthis). All donations go to our bounty fund and allow us to place bounties on important bugs and enhancements. You may also use Beerpay linked via the above badges.
|
||||
|
||||
|
||||
@@ -9,7 +9,7 @@ class AMQMessageFactoryTest(TestCase):
|
||||
raw = "\002\027A64f3b5af4d00000a4000|m=POST|d=clever-pybook.localhost|p=/debug|h={Accept=[*/*],X-Forwarded-Host=[clever-pybook.localhost:8085],User-Agent=[curl/8.7.1],X-Forwarded-Proto=[http],X-Forwarded-For=[172.18.0.1],Host=[clever-pybook.localhost:8085],Accept-Encoding=[gzip],Content-Length=[1475],X-Forwarded-Port=[8085],X-Real-Ip=[172.18.0.1],X-Forwarded-Server=[c5945bef67d1],Content-Type=[multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v]}|t={traceparent=00-bed6eb7de0d8ccf4c80b145a4bfec357-8d1f3bafc471bf0f-01}|b=TEST_ME"
|
||||
msg = AMQMessageFactory.from_bytes(raw.encode("utf-8"))
|
||||
self.assertIsNotNone(msg.id)
|
||||
self.assertEqual(msg.id.as_string(),'64F3B5AF4D00000A4000','SnowflakeID incorrectly parsed.')
|
||||
self.assertEqual(msg.id.as_string(),'64F3B5AF4D00000A4000'.lower(),'SnowflakeID incorrectly parsed.')
|
||||
self.assertEqual(msg.headers["Content-Type"][0], "multipart/form-data; boundary=------------------------j1RjIqLrFH07XMsyMvpS1v")
|
||||
|
||||
def test_generate_snowflake_id(self):
|
||||
@@ -24,10 +24,10 @@ class AMQMessageFactoryTest(TestCase):
|
||||
_i1 = _f.generate_snowflake_id()
|
||||
_i2 = _f.generate_snowflake_id()
|
||||
_i3 = _f.generate_snowflake_id()
|
||||
self.assertEqual('0007F000', _i0.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
self.assertEqual('0007F001', _i1.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
self.assertEqual('0007F002', _i2.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
self.assertEqual('0007F003', _i3.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
self.assertEqual('0007F000'.lower(), _i0.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
self.assertEqual('0007F001'.lower(), _i1.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
self.assertEqual('0007F002'.lower(), _i2.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
self.assertEqual('0007F003'.lower(), _i3.as_string()[-8:], 'Instance ID and sequence not set correctly')
|
||||
|
||||
def test_create_request_message(self):
|
||||
_f = AMQMessageFactory(34)
|
||||
@@ -65,4 +65,4 @@ class AMQMessageFactoryTest(TestCase):
|
||||
_f = AMQMessageFactory(35)
|
||||
_req = _f.create_request_message("PUT", "localhost", "/test/me", {"k1": ["v1"], "k2": ["v2"]}, "Quick brown fox")
|
||||
_as_str = _f.to_string(_req)
|
||||
self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found")
|
||||
self.assertGreater(_as_str.index("domain='localhost'"), 0, "Expected content not found")
|
||||
|
||||
@@ -6,7 +6,6 @@ from asyncio import Future
|
||||
|
||||
from amqp.adapter.amq_message_factory import AMQMessageFactory
|
||||
from amqp.adapter.amq_route_factory import AMQRouteFactory
|
||||
from cleverswarm.clever_swarm_adapter import CleverSwarmAdapter
|
||||
from amqp.adapter.service_message_factory import CleverMicroServiceMessageFactory
|
||||
|
||||
from amqp.config.amq_configuration import AMQConfiguration
|
||||
@@ -87,7 +86,7 @@ class AMQService:
|
||||
|
||||
if __name__ == "__main__":
|
||||
amq_configuration: AMQConfiguration = AMQConfiguration()
|
||||
amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAdapter(amq_configuration))
|
||||
amq_service.rabbit_mq_consumer.channel.start_consuming()
|
||||
# amq_service: AMQService = AMQService(amq_configuration, CleverSwarmAmqpAdapter(amq_configuration))
|
||||
# amq_service.rabbit_mq_consumer.channel.start_consuming()
|
||||
# amq_service.consumer_thread.cancel()
|
||||
# amq_service.rabbit_mq_consumer.connection.close()
|
||||
|
||||
@@ -13,7 +13,7 @@ from pika.channel import Channel
|
||||
|
||||
from amqp.adapter.amq_message_factory import AMQMessageFactory
|
||||
from amqp.adapter.amq_response_factory import AMQResponseFactory
|
||||
from cleverswarm.clever_swarm_adapter import CleverSwarmAdapter
|
||||
from cleverswarm.clever_swarm_adapter import CleverSwarmAmqpAdapter
|
||||
from amqp.model.model import AMQResponse, AMQMessage
|
||||
from amqp.router.router_base import AMQ_REPLY_TO_EXCHANGE
|
||||
from amqp.router.serializer import map_as_string
|
||||
@@ -37,8 +37,8 @@ def set_text(carrier, key, value):
|
||||
|
||||
|
||||
class AMQServiceMessageHandler:
|
||||
def __init__(self, service_adapter: CleverSwarmAdapter, tracer: Tracer, message_factory: AMQMessageFactory):
|
||||
self.http_adapter: CleverSwarmAdapter = service_adapter
|
||||
def __init__(self, service_adapter: CleverSwarmAmqpAdapter, tracer: Tracer, message_factory: AMQMessageFactory):
|
||||
self.http_adapter: CleverSwarmAmqpAdapter = service_adapter
|
||||
self.tracer = tracer
|
||||
self.message_factory: AMQMessageFactory = message_factory
|
||||
# Dictionary to store outstanding Futures
|
||||
|
||||
@@ -1,226 +0,0 @@
|
||||
import logging
|
||||
from typing import Tuple, AnyStr, Any
|
||||
from aiohttp import ClientSession, ClientResponse
|
||||
|
||||
import cleverswarm.core.deps
|
||||
from amqp.adapter.amq_response_factory import AMQResponseFactory
|
||||
from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, CleverMultiPartParser, \
|
||||
call_cleverswarm_get_api, call_cleverswarm_post_put_api
|
||||
from amqp.config.amq_configuration import AMQConfiguration
|
||||
from amqp.model.model import AMQMessage, AMQResponse
|
||||
|
||||
import cleverswarm.rest.v0.endpoints.benchmark as benchmark
|
||||
import cleverswarm.rest.v0.endpoints.jobs as jobs
|
||||
import cleverswarm.rest.v0.endpoints.unstructured as unstructured
|
||||
from cleverswarm.schemas.user_schema import UserSchema
|
||||
|
||||
|
||||
BENCHMARK = '/benchmark'
|
||||
JOBS = '/jobs/'
|
||||
UNSTRUCTURED = '/unstructured'
|
||||
|
||||
GET_BENCHMARK_KG = r".*/benchmark/(?P<request_id>[^\?]+)\?index=(?P<index>\d+)"
|
||||
GET_JOB_STATUS = r".*/jobs/(?P<request_id>.*)"
|
||||
GET_KG = r".*/unstructured/(?P<request_id>.*)"
|
||||
DUMMY = r"(?P<request_id>.*)"
|
||||
|
||||
#
|
||||
# Following methods are wrappers to actual CleverSwarm endpoints, so I can refer to the CleverSwarm method and
|
||||
# pass it as input parameter using general Callable[[Any, UserSchema], str] type
|
||||
#
|
||||
async def cleverswarm_create_benchmark_job(args: Any, authenticated_user: UserSchema) -> str:
|
||||
return str(await benchmark.create_benchmark_job(**args, authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_create_extract_with_ontology(args: Any, authenticated_user: UserSchema) -> str:
|
||||
return str(await unstructured.create_extract_with_ontology(**args, authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_create_extract_with_wildcards(args: Any, authenticated_user: UserSchema) -> str:
|
||||
return str(await unstructured.create_extract_with_ontology(**args, authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_update_benchmark_job(args: Any, authenticated_user: UserSchema) -> str:
|
||||
return str(await benchmark.update_benchmark_job(**args, authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_retry_job(args: Any, authenticated_user: UserSchema) -> str:
|
||||
return str(await jobs.retry_job(args[0], authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_delete_job(args: Any, authenticated_user: UserSchema) -> str:
|
||||
return str(await jobs.delete_job(args[0], authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_get_benchmark_kg(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str:
|
||||
return str(await benchmark.get_benchmark_kg(args[0], args[1], authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_get_job_status(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str:
|
||||
return str(await jobs.get_job_status(args[0], authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_jobs_get_requests(args: Tuple[AnyStr | Any, ...],
|
||||
authenticated_user: UserSchema) -> str:
|
||||
return str(await jobs.get_requests(authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
async def cleverswarm_get_kg(args: Tuple[AnyStr | Any, ...], authenticated_user: UserSchema) -> str:
|
||||
return str(await unstructured.get_kg(args[0], authenticated_user=authenticated_user))
|
||||
|
||||
|
||||
class CleverSwarmAdapter(CleverThisServiceAdapter):
|
||||
def __init__(self, amq_configuration: AMQConfiguration, session: ClientSession = None):
|
||||
super().__init__(amq_configuration, session)
|
||||
|
||||
async def get_current_user(self, token: str) -> UserSchema:
|
||||
return await cleverswarm.core.deps.get_current_user(token)
|
||||
# ==================================================================================================
|
||||
# ===================== C l e v e r S w a r m A P I m e t h o d s ============================
|
||||
# ==================================================================================================
|
||||
|
||||
async def get(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
|
||||
"""
|
||||
Handles the GET requests for the CleverSwarm API.
|
||||
:param message: message from the AMQP that contains the GET request
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
if BENCHMARK in message.path:
|
||||
return await call_cleverswarm_get_api(
|
||||
message,
|
||||
GET_BENCHMARK_KG,
|
||||
cleverswarm_get_benchmark_kg,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
if JOBS in message.path:
|
||||
return await call_cleverswarm_get_api(
|
||||
message,
|
||||
GET_JOB_STATUS,
|
||||
cleverswarm_get_job_status,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
if message.path.endswith('/jobs'):
|
||||
return await call_cleverswarm_get_api(
|
||||
message,
|
||||
DUMMY,
|
||||
cleverswarm_jobs_get_requests,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
if UNSTRUCTURED in message.path:
|
||||
return await call_cleverswarm_get_api(
|
||||
message,
|
||||
GET_KG,
|
||||
cleverswarm_get_kg,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
# unmatched path - try to invoke the service directly on this path
|
||||
url = f"http://{self.service_host}:{self.service_port}{message.path}"
|
||||
headers = {**message.headers, **message.trace_info}
|
||||
logging.info(f" [*] SEND HEADERS out: {', '.join(f'{k}={v}' for k, v in headers.items())}, REQ={url}")
|
||||
if self.session:
|
||||
_http_resp: ClientResponse = await self.session.get(url, headers=headers)
|
||||
return AMQResponseFactory.of(message.id, _http_resp.status, _http_resp.content_type,
|
||||
await _http_resp.read(),
|
||||
message.trace_info)
|
||||
return AMQResponseFactory.of(message.id, 404, "text/plain",
|
||||
str("Destination location does not exists").encode('utf-8'), message.trace_info)
|
||||
|
||||
async def put(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
|
||||
"""
|
||||
Handles the PUT requests for the CleverSwarm API.
|
||||
:param message: message from the AMQP that contains the PUT request
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
parser: CleverMultiPartParser = CleverMultiPartParser(message)
|
||||
if BENCHMARK in message.path:
|
||||
return await call_cleverswarm_post_put_api(
|
||||
message,
|
||||
parser.form_data,
|
||||
cleverswarm_create_benchmark_job,
|
||||
201,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
if JOBS in message.path:
|
||||
return await call_cleverswarm_get_api(
|
||||
message,
|
||||
GET_JOB_STATUS,
|
||||
cleverswarm_retry_job,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
return await self.handle_possible_form(
|
||||
self.session.put(
|
||||
f"http://{self.service_host}:{self.service_port}{message.path}",
|
||||
data=message.body,
|
||||
headers={**message.headers, **message.trace_info},
|
||||
),
|
||||
auth_user
|
||||
)
|
||||
|
||||
async def post(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
|
||||
"""
|
||||
Handles the POST requests for the CleverSwarm API.
|
||||
:param message: message from the AMQP that contains the POST request, including body and headers
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
parser: CleverMultiPartParser = CleverMultiPartParser(message)
|
||||
if BENCHMARK in message.path:
|
||||
return await call_cleverswarm_post_put_api(
|
||||
message,
|
||||
parser.form_data,
|
||||
cleverswarm_create_benchmark_job,
|
||||
201,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
if UNSTRUCTURED in message.path:
|
||||
if 'with_ontology' in message.path:
|
||||
return await call_cleverswarm_post_put_api(
|
||||
message,
|
||||
parser.form_data,
|
||||
cleverswarm_create_extract_with_ontology,
|
||||
200,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
if 'with_wildcards' in message.path:
|
||||
return await call_cleverswarm_post_put_api(
|
||||
message,
|
||||
parser.form_data,
|
||||
cleverswarm_create_extract_with_wildcards,
|
||||
200,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
return await self.handle_possible_form(
|
||||
self.session.post(
|
||||
f"http://{self.service_host}:{self.service_port}{message.path}",
|
||||
data=message.body,
|
||||
headers={**message.headers, **message.trace_info},
|
||||
),
|
||||
auth_user
|
||||
)
|
||||
|
||||
async def handle_possible_form(self, request_coroutine, auth_user: UserSchema) -> AMQResponse:
|
||||
return await request_coroutine
|
||||
|
||||
async def head(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
|
||||
raise NotImplementedError
|
||||
|
||||
async def delete(self, message: AMQMessage, auth_user: UserSchema) -> AMQResponse:
|
||||
parser: CleverMultiPartParser = CleverMultiPartParser(message)
|
||||
if JOBS in message.path:
|
||||
return await call_cleverswarm_get_api(
|
||||
message,
|
||||
GET_JOB_STATUS,
|
||||
cleverswarm_retry_job,
|
||||
authenticated_user=auth_user
|
||||
)
|
||||
|
||||
raise NotImplementedError
|
||||
@@ -1,58 +0,0 @@
|
||||
from pytz import timezone
|
||||
|
||||
from typing import Optional
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from fastapi.security import OAuth2PasswordBearer
|
||||
|
||||
import jwt
|
||||
|
||||
from cleverswarm.core.configs import settings
|
||||
from cleverswarm.core.security import validate_pass, create_password_hash
|
||||
from cleverswarm.schemas.user_schema import UserSchema
|
||||
|
||||
oauth2_schema = OAuth2PasswordBearer(
|
||||
tokenUrl=f"{settings.API_V0_STR}/login"
|
||||
)
|
||||
|
||||
|
||||
async def authenticate(username: str, password: str) -> Optional[UserSchema]:
|
||||
if username != 'user':
|
||||
return None
|
||||
|
||||
test_hash = create_password_hash('user1234')
|
||||
|
||||
if not validate_pass(password, test_hash):
|
||||
return None
|
||||
|
||||
# The returned value at this point should be UserModel, but since we do not have any persistence
|
||||
# and this is for demo purposes only
|
||||
return UserSchema(id=0, username='user', hash_pass=test_hash)
|
||||
|
||||
def _create_token(token_type: str, lifetime: timedelta, sub: str) -> str:
|
||||
# https://dataracker.ietf.org/doc/html/rfc7519#section-4.1.3
|
||||
payload = {}
|
||||
|
||||
pt = timezone('UTC')
|
||||
expires = datetime.now(tz=pt) + lifetime
|
||||
|
||||
payload["type"] = token_type
|
||||
|
||||
payload["exp"] = expires
|
||||
|
||||
payload["iat"] = datetime.now(tz=pt) #Issued-At (iat)
|
||||
|
||||
payload["sub"] = str(sub)
|
||||
|
||||
return jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.ALGORITHM)
|
||||
|
||||
|
||||
def create_access_token(sub: str) -> str:
|
||||
"""
|
||||
https://jwt.io
|
||||
"""
|
||||
return _create_token(
|
||||
token_type='access_token',
|
||||
lifetime=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES),
|
||||
sub=sub
|
||||
)
|
||||
@@ -1,28 +0,0 @@
|
||||
from pydantic_settings import BaseSettings
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
API_V0_STR: str = '/api/v0'
|
||||
|
||||
# JSON Web Tokens (https://jwt.io) - Secret utilized for token generation
|
||||
JWT_SECRET: str = 'qS96E1oCfq5gEZH-ngD91NC2qkcl0cffhNTIDGpF4pw'
|
||||
# Token was generated by the code below:
|
||||
"""
|
||||
import secrets
|
||||
|
||||
token: str = secrets.token_urlsafe(32) # com 32 caracteres
|
||||
"""
|
||||
ALGORITHM: str = 'HS256'
|
||||
# Token timeout with 1h
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60
|
||||
|
||||
JOBS_FILE_PATH: Path = 'jobs'
|
||||
|
||||
BASE_PATH: Path = ''
|
||||
|
||||
class Config:
|
||||
case_sensitive = True
|
||||
|
||||
settings: Settings = Settings()
|
||||
|
||||
@@ -1,60 +0,0 @@
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import Depends, HTTPException, status
|
||||
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||||
|
||||
#from jose import jwt, JWTError
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
from cleverswarm.schemas.user_schema import UserSchema
|
||||
from cleverswarm.core.auth import oauth2_schema, authenticate
|
||||
from cleverswarm.core.security import create_password_hash
|
||||
|
||||
class TokenData(BaseModel):
|
||||
userid: Optional[str] = None
|
||||
|
||||
async def get_current_user(
|
||||
token: str = Depends(oauth2_schema)
|
||||
) -> UserSchema:
|
||||
credential_exception: HTTPException = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail='User provided credentials are not correct.',
|
||||
headers={"WWW-Authenticate": "Bearer"}
|
||||
)
|
||||
|
||||
try:
|
||||
#payload = jwt.decode(
|
||||
# token,
|
||||
# settings.JWT_SECRET,
|
||||
# algorithms=[settings.ALGORITHM],
|
||||
# options={"verify_aud": False}
|
||||
#)
|
||||
userid: str = 'user' #payload.get("sub")
|
||||
if userid is None:
|
||||
raise credential_exception
|
||||
|
||||
token_data: TokenData = TokenData(userid=userid)
|
||||
except Exception:
|
||||
raise credential_exception
|
||||
|
||||
if int(token_data.userid) != 0:
|
||||
raise credential_exception
|
||||
|
||||
# The returned value at this point should be UserModel, but since we do not have any persistence
|
||||
# and this is for demo purposes only
|
||||
return UserSchema(id=token_data.userid, username='user', hash_pass=create_password_hash('user1234'))
|
||||
|
||||
|
||||
security = HTTPBasic()
|
||||
|
||||
|
||||
async def get_current_username(credentials: HTTPBasicCredentials = Depends(security)) -> str:
|
||||
user = await authenticate(credentials.username, credentials.password)
|
||||
if user is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Incorrect username or password",
|
||||
headers={"WWW-Authenticate": "Basic"},
|
||||
)
|
||||
return credentials.username
|
||||
@@ -1,20 +0,0 @@
|
||||
from passlib.context import CryptContext
|
||||
|
||||
CRIPTO = CryptContext(schemes=['bcrypt'], deprecated='auto') # Automatically workaround deprecated APIs
|
||||
|
||||
|
||||
def validate_pass(password: str, hash_pass: str) -> bool:
|
||||
"""
|
||||
Verifies if a given password matches the expected hash.
|
||||
It compares the user provided text password with the stored hash of the expected password, that
|
||||
is currently stored in the system.
|
||||
"""
|
||||
|
||||
return CRIPTO.verify(password, hash_pass)
|
||||
|
||||
|
||||
def create_password_hash(password: str) -> str:
|
||||
"""
|
||||
Return the hash of the password
|
||||
"""
|
||||
return CRIPTO.hash(password)
|
||||
@@ -1,111 +0,0 @@
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, status, Depends
|
||||
from fastapi import Path as FPath
|
||||
from fastapi import Query
|
||||
from fastapi import UploadFile
|
||||
from fastapi.responses import Response
|
||||
|
||||
from cleverswarm.schemas.request_schema import RequestSchema
|
||||
from cleverswarm.schemas.user_schema import UserSchema
|
||||
from cleverswarm.schemas.job_id_schema import JobIdSchema
|
||||
|
||||
from cleverswarm.core.deps import get_current_user
|
||||
|
||||
logger = logging.getLogger('Benchmark')
|
||||
# Enable propagation to the root logger
|
||||
logger.propagate = True
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# POST Convert
|
||||
@router.post('/',
|
||||
status_code=status.HTTP_201_CREATED, summary='Creates a new benchmark job',
|
||||
description='Creates a new benchmark job/request, however if additional input data files '
|
||||
'are required they must be provided through an addition PUT method call. '
|
||||
'When all the job files have been uploaded a PUT method call must be invoked to'
|
||||
'set the job/request status to ReadyForProcessing.',
|
||||
responses={
|
||||
201: {'model': JobIdSchema,
|
||||
'description': 'Job updated with new input files'},
|
||||
507: {'description': 'The server possibly ran out of disk space or an internal server error occurred'}
|
||||
})
|
||||
async def create_benchmark_job(unstructured: UploadFile,
|
||||
ontology: UploadFile,
|
||||
ground_truth: UploadFile,
|
||||
ready_for_processing: bool = Query(
|
||||
title='Defines if the job is ready for processing',
|
||||
description='If set to true the job will be marked as ready '
|
||||
'to start being processed. No more files can uploaded '
|
||||
'for this benchmark job, after being marked as ready',
|
||||
examples=[True, False]),
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logging.info('===> POST /benchmark: Started Benchmark job creation')
|
||||
logging.info('===> POST arguments: unstructured=%s, ontology=%s, ground_truth=%s, ready_for_processing=%s',
|
||||
unstructured, ontology, ground_truth, ready_for_processing)
|
||||
return JobIdSchema(job_id='123456-TestKG')
|
||||
|
||||
|
||||
@router.put('/{job_id}', status_code=status.HTTP_202_ACCEPTED, response_model=RequestSchema,
|
||||
summary='Append additional files to the benchmark request.',
|
||||
description='Append additional files to the benchmark request. Only possible if Job/Request '
|
||||
'is still in the Created state.',
|
||||
responses={
|
||||
202: {'model': RequestSchema,
|
||||
'description': 'Job updated with new input files'},
|
||||
400: {'description': 'Job does not correspond to a Benchmark'},
|
||||
404: {'description': 'Job ID not found or invalid index'},
|
||||
412: {'description': 'Job is in state that does not allow it to be updated.'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def update_benchmark_job(unstructured: UploadFile,
|
||||
ontology: UploadFile,
|
||||
ground_truth: UploadFile,
|
||||
job_id: str = FPath(
|
||||
title='The Job/Request identifier',
|
||||
description='The identifier string for the desired Job/Request '
|
||||
'to be updated with additional files for processing',
|
||||
example='006ea377-baf1-4e04-b18b-37781578d81a'),
|
||||
ready_for_processing: bool = Query(
|
||||
title='Defines if the job is ready for processing',
|
||||
description='If set to true the job will be marked as ready '
|
||||
'to start being processed. No more files can uploaded '
|
||||
'for this benchmark job, after being marked as ready',
|
||||
examples=[True, False]),
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
|
||||
logging.info(f'===> PUT /benchmark/{job_id}: Append additional files to the benchmark request')
|
||||
logging.info('===> PUT arguments: unstructured=%s, ontology=%s, ground_truth=%s, ready_for_processing=%s',
|
||||
unstructured, ontology, ground_truth, ready_for_processing)
|
||||
return RequestSchema(job_id=job_id, status='Updated')
|
||||
|
||||
class JSONLResponse(Response):
|
||||
media_type = "application/jsonl"
|
||||
|
||||
|
||||
@router.get('/{request_id}', summary='Polls or Retrieves a benchmark KG by id and index',
|
||||
description='Polls or Retrieves a processed benchmark KG by id and index. '
|
||||
'If the KG is not ready yet it sends a NO CONTENT (204) HTTP status.',
|
||||
response_class=JSONLResponse,
|
||||
response_description='Response is in the form of a jsonl file',
|
||||
responses={
|
||||
200: {'content': {'c': {}},
|
||||
'description': 'The benchmark KG extracted data'},
|
||||
204: {'description': 'Job is not completed yet'},
|
||||
404: {'description': 'Job ID not found or invalid index'},
|
||||
412: {'description': 'Job does not correspond to a Benchmark'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def get_benchmark_kg(request_id: str =
|
||||
FPath(title='The request id',
|
||||
description='The id of the request for which the results are to be retrieved',
|
||||
example='3286f418-8141-450f-8ad5-f6a5c01a0ce9'),
|
||||
index: int = Query(title='The index of the result to be retrieved',
|
||||
description='The index of the result to retrieved starting at 0 and '
|
||||
'up to the number of benchmark files -1.',
|
||||
ge=0,
|
||||
example=0),
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logging.info(f'===> GET /benchmark/{request_id}?index={index}: Polls or Retrieves a benchmark KG by id and index')
|
||||
return JSONLResponse(content=f'{"index": {index}, "request_id": "{request_id}", "data": "data"}')
|
||||
@@ -1,74 +0,0 @@
|
||||
from typing import List
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, status, Depends
|
||||
from fastapi import Path as FPath
|
||||
|
||||
from cleverswarm.schemas.request_schema import RequestSchema
|
||||
from cleverswarm.schemas.user_schema import UserSchema
|
||||
|
||||
from cleverswarm.core.deps import get_current_user
|
||||
|
||||
logger = logging.getLogger('Jobs')
|
||||
# Enable propagation to the root logger
|
||||
logger.propagate = True
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.get('/{request_id}', summary='Obtains the status of a given job',
|
||||
description='Obtains the status of an existing job by its id',
|
||||
status_code=status.HTTP_200_OK,
|
||||
responses={
|
||||
200: {'description': 'Job status retrieved successfully'},
|
||||
404: {'description': 'Job not found'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def get_job_status(request_id: str = FPath(title='Job ID',
|
||||
description='The id of the job for which the status is to be queried'),
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logging.info(f'===> GET /jobs/{request_id}: Obtains the status of a given job')
|
||||
return "JobStatus.Completed"
|
||||
|
||||
@router.get('/', summary='List all user submitted requests',
|
||||
description='List all user submitted requests along with their status',
|
||||
status_code=status.HTTP_200_OK,
|
||||
response_model=List[RequestSchema],
|
||||
response_description='User submitted job requests',
|
||||
responses={
|
||||
200: {'description': 'User submitted job requests'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def get_requests(authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logging.info('===> GET /jobs: List all user submitted requests')
|
||||
return []
|
||||
|
||||
|
||||
@router.delete('/{request_id}', summary='Deletes a completed or failed job',
|
||||
description='Deletes a completed or failed job by its request_id',
|
||||
status_code=status.HTTP_204_NO_CONTENT,
|
||||
responses={
|
||||
204: {'description': 'Job data was successfully deleted'},
|
||||
404: {'description': 'Job ID not found'},
|
||||
412: {'description': 'Job is in state that does not allow deletion'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def delete_job(request_id: str = FPath(title='Request ID',
|
||||
description='The id of the failed or completed request to be deleted'),
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logging.info(f'===> DELETE /jobs/{request_id}: Deletes a completed or failed job')
|
||||
return 204
|
||||
|
||||
@router.put('/{request_id}', summary='Retries execution of a given job',
|
||||
description='Retries the execution of a given job by its id, provided the current job status is failed',
|
||||
status_code=status.HTTP_202_ACCEPTED,
|
||||
responses={
|
||||
202: {'model': RequestSchema,
|
||||
'description': 'The updated job status and its configuration'},
|
||||
406: {'description': 'Job is an state that does not allow it to be retried'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def retry_job(request_id: str = FPath(title='Job ID',
|
||||
description='The id of the job to be retried'),
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logging.info(f'===> PUT /jobs/{request_id}: Retries execution of a given job')
|
||||
return RequestSchema(job_id=request_id, status='Failed')
|
||||
@@ -1,77 +0,0 @@
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, status, Depends
|
||||
from fastapi import Path as FPath
|
||||
from fastapi import UploadFile
|
||||
|
||||
from cleverswarm.schemas.user_schema import UserSchema
|
||||
from cleverswarm.schemas.job_id_schema import JobIdSchema
|
||||
|
||||
from cleverswarm.core.deps import get_current_user
|
||||
|
||||
logger = logging.getLogger('Unstructured')
|
||||
# Enable propagation to the root logger
|
||||
logger.propagate = True
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# POST Convert
|
||||
@router.post('/with_ontology', status_code=status.HTTP_201_CREATED, summary='Convert unstructured text to KG',
|
||||
description='Converts unstructured text to a Knowledge Graph according to the provided ontology',
|
||||
response_description='Returns the Job/Request ID for reference in future interactions',
|
||||
responses={
|
||||
200: {'model': JobIdSchema,
|
||||
'description': 'The KG extraction job ID'},
|
||||
400: {'description': 'Input parameters are not compliant with request'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def create_extract_with_ontology(unstructured: UploadFile,
|
||||
ontology: UploadFile,
|
||||
ontology_spec: UploadFile,
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logger.debug('====> POST /unstructured/with_ontology: Started convertToKG')
|
||||
logger.debug('====> arguments: unstructured=%s, ontology=%s, ontology_spec=%s', unstructured, ontology, ontology_spec)
|
||||
return JobIdSchema(job_id='123456-TestKG')
|
||||
|
||||
|
||||
# POST Convert
|
||||
@router.post('/with_wildcards', status_code=status.HTTP_201_CREATED,
|
||||
summary='Convert unstructured text to KG with Wildcards filtering',
|
||||
description='Converts unstructured text to a Knowledge Graph according to the provided ontology, while'
|
||||
'applying a filtering based on the Wildcards ontology',
|
||||
response_description='Knowledge Graph defined by RDF triplets in JSON format',
|
||||
responses={
|
||||
200: {'model': JobIdSchema,
|
||||
'description': 'The KG extraction job ID'},
|
||||
400: {'description': 'Input parameters are not compliant with request'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def create_extract_with_ontology(unstructured: UploadFile,
|
||||
ontology: UploadFile,
|
||||
ontology_spec: UploadFile,
|
||||
wildcards: UploadFile,
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logger.debug('====> POST /unstructured/with_wildcards: Started convertToKG with Wildcards filtering')
|
||||
logger.debug('====> arguments: unstructured=%s, ontology=%s, ontology_spec=%s, wildcards=%s',
|
||||
unstructured, ontology, ontology_spec, wildcards)
|
||||
return JobIdSchema(job_id='123456-TestKG')
|
||||
|
||||
|
||||
@router.get('/{request_id}', summary='Polls or Retrieves a KG by id',
|
||||
description='Polls or Retrieves a processed KG by id. If the KG is not ready yet it sends a '
|
||||
'NO CONTENT (204) HTTP status',
|
||||
response_description='Response is in the form of json data',
|
||||
responses={
|
||||
200: {'description': 'The KG extracted data'},
|
||||
204: {'description': 'Job is not completed yet'},
|
||||
404: {'description': 'Job ID not found'},
|
||||
412: {'description': 'Job does not correspond to a KG extraction job'},
|
||||
500: {'description': 'A server error occurred while processing the request'}
|
||||
})
|
||||
async def get_kg(request_id: str = FPath(title='The job id',
|
||||
description='The id of the job for which the results are to be retrieved'),
|
||||
authenticated_user: UserSchema = Depends(get_current_user)):
|
||||
logger.debug(f'====> GET /unstructured/{request_id}: Started convertToKG with Wildcards filtering')
|
||||
|
||||
|
||||
@@ -1,5 +0,0 @@
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class JobIdSchema(BaseModel):
|
||||
job_id: str
|
||||
@@ -1,17 +0,0 @@
|
||||
from typing import Optional, List
|
||||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class RequestSchema(BaseModel):
|
||||
id: str = None
|
||||
created: datetime
|
||||
type: str
|
||||
status: str
|
||||
retries_count: int
|
||||
unstructured_sources: List[str]
|
||||
ontology_sources: List[str]
|
||||
ontology_spec_sources: List[str]
|
||||
wildcards_sources: List[str]
|
||||
ground_truth_sources: List[str]
|
||||
@@ -1,9 +0,0 @@
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class UserSchema(BaseModel):
|
||||
id: Optional[int] = None
|
||||
username: str
|
||||
hash_pass: str
|
||||
@@ -0,0 +1,20 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=61.0", "wheel"] # Specifies build tools to use
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "amq_adapter"
|
||||
version = "0.1.0"
|
||||
description = "The Python implementation of the AMQ Adaptor for CleverMicro"
|
||||
readme = "README.md"
|
||||
license = { text = "Apache License 2.0" }
|
||||
dependencies = [
|
||||
"aiohttp~=3.11.11",
|
||||
"pika~=1.3.2",
|
||||
"fastapi==0.115.6",
|
||||
"opentelemetry-api",
|
||||
"opentelemetry-sdk",
|
||||
"opentelemetry-exporter-otlp",
|
||||
"opentelemetry-exporter-prometheus",
|
||||
"opentelemetry-instrumentation-pika"
|
||||
]
|
||||
Reference in New Issue
Block a user