Open Discover Service

FROM python:3-alpine

FROM labs.etsi.org:5050/ocf/capif/python:3-slim-bullseye

RUN mkdir -p /usr/src/app

WORKDIR /usr/src/app

COPY requirements.txt /usr/src/app/

COPY requirements.txt /usr/src/app/requirements.txt

RUN apt-get update && apt-get install -y --no-install-recommends \

    gcc \

    build-essential \

    linux-headers-generic \

    ca-certificates \

    libffi-dev \

    libssl-dev \

    libxslt1-dev \

    jq \

    curl && \

    rm -rf /var/lib/apt/lists/*

RUN pip3 install --no-cache-dir -r requirements.txt

EXPOSE 8080

ENTRYPOINT ["python3"]

CMD ["-m", "openapi_server"]

 No newline at end of file

CMD ["sh", "prepare_open_discover.sh"]

mongo: {

  'user': 'root',

  'password': 'example',

  'db': 'capif',

  'col': 'serviceapidescriptions',

  'host': 'mongo',

  'port': "27017"

}

#!/usr/bin/env python3

import connexion

from openapi_server import encoder

def main():

    app = connexion.App(__name__, specification_dir='./openapi/')

    app.app.json_encoder = encoder.JSONEncoder

    app.add_api('openapi.yaml',

                arguments={'title': 'CAPIF_Open_Discover_Service_API'},

                pythonic_params=True)

    app.run(port=8080)

if __name__ == '__main__':

    main()

#!/usr/bin/env python3

import json

import connexion

import encoder

from connexion.decorators import parameter as connexion_parameter

from flask_jwt_extended import JWTManager

_original_get_val_from_param = connexion_parameter._get_val_from_param

def _get_val_from_content_param(value, param_defn):

    if "content" not in param_defn or "schema" in param_defn:

        return _original_get_val_from_param(value, param_defn)

    content = param_defn.get("content", {})

    if "application/json" not in content or not isinstance(value, str):

        return value

    try:

        return json.loads(value)

    except json.JSONDecodeError:

        return value

connexion_parameter._get_val_from_param = _get_val_from_content_param

with open("/usr/src/app/openapi_server/pubkey.pem", "rb") as pub_file:

    pub_data = pub_file.read()

app = connexion.App(__name__, specification_dir="openapi/")

app.app.json_encoder = encoder.CustomJSONEncoder

app.add_api(

    "openapi.yaml",

    arguments={"title": "CAPIF_Open_Discover_Service_API"},

    pythonic_params=True,

)

app.app.config["JWT_ALGORITHM"] = "RS256"

app.app.config["JWT_PUBLIC_KEY"] = pub_data

JWTManager(app.app)

import os

import yaml

class Config:

    def __init__(self):

        self.cached = 0

        self.file = "../config.yaml"

        self.my_config = {}

        stamp = os.stat(self.file).st_mtime

        if stamp != self.cached:

            self.cached = stamp

            with open(self.file) as f:

                self.my_config = yaml.safe_load(f)

    def get_config(self):

        return self.my_config