D-Net

certifi==2019.3.9

chardet==3.0.4

Click==7.0

elasticsearch==5.5.3

elasticsearch-dsl==5.4.0

fastapi==0.20.0

h11==0.8.1

httptools==0.0.13

idna==2.8

memory-profiler==0.55.0

psutil==5.6.2

pydantic==0.23

Pympler==0.7

python-dateutil==2.8.0

requests==2.22.0

six==1.12.0

starlette==0.11.1

urllib3==1.25.2

uvicorn==0.7.1

uvloop==0.12.2

websockets==7.0

prometheus-client==0.5.0

from fastapi import FastAPI, Query,HTTPException

from v1.responseModel import ScholixType,convert_response

import logging

from typing import List

from eu.dnetlib.ScholixConnector import ScholixConnector

import logging

log = logging.getLogger("scholexplorer")

subapi_v1 = FastAPI(title="Scholexplorer API 1.0",

    version="2.0.0",

    description="scholexplorer API version 1.0",openapi_prefix="/v1", docs_url="/ui/")

@subapi_v1.get("/linksFromDatasource", 

        tags=["Scholix"], 

        summary="Get all scholix relation collected from a datasource",     

        description="return a list of scholix object collected from a specific datasource" ,

        response_model=List[ScholixType]        

        )

def links_from_datasource(datasource: str = Query((...),title="datasource", description="Filter Scholix relationships collected from a LinkProvider"),page:int= Query(None, title="page", description="select page of result")):

    s = ScholixConnector()

    m_page = 0

    if page:

            m_page= page*100      

    if m_page > 9999:

        raise HTTPException(status_code=400,detail="MAX NUMBER OF PAGE REACHED")  

    return convert_response(s.links(provider=datasource, page= m_page))

@subapi_v1.get("/linksFromPid", 

        tags=["Scholix"], 

        summary="Retrieve all scholix links from a persistent identifier",     

        description="The linksFromPid endpoint returns a list of scholix object related from a specific persistent identifier" ,

        response_model=List[ScholixType]        

        )

def links_from_pid(

    pid: str = Query((...), title="pid", description="persistent Identifier"),

    pidType: str = Query(None, title="pidType", description="Persistent Identifier Type"),

    typologyTarget: str = Query(None, title="targetPidType", description="typology target filter should be publication, dataset or unknown"),

    datasourceTarget: str = Query(None, title="datasourceTarget", description="a datasource provenace filter of the target relation"),

    page:int= Query(None, title="page", description="select page of result")):

    s = ScholixConnector()    

    m_page = 0

    if page:

            m_page= page*100      

    if m_page > 9999:

        raise HTTPException(status_code=400,detail="MAX NUMBER OF PAGE REACHED")  

    return convert_response(s.links(s_pid =pid,s_pid_type= pidType,source_Type= typologyTarget, provider = datasourceTarget) )

@subapi_v1.get("/linksFromPublisher", 

        tags=["Scholix"], 

        summary="Get all scholix relation collected from a publisher",     

        description="return a list of scholix object published from a specific publisher" ,

        response_model=List[ScholixType]        

        )

def links_from_publisher(publisher: str = Query((...),title="publisher", description="Filter Scholix relationships collected from a publisher"),page:int= Query(None, title="page", description="select page of result")):

    s = ScholixConnector()

    m_page = 0

    if page:

            m_page= page*100      

    if m_page > 9999:

        raise HTTPException(status_code=400,detail="MAX NUMBER OF PAGE REACHED")  

    return convert_response(s.links(t_publisher=publisher, page= m_page))

@subapi_v1.get("/listDatasources", 

        tags=["Datasources"], 

        summary="Get all datasources",     

        description="returns a list of all datasources" ,

        response_model=List[str]        

        )

def list_datasources():

    s = ScholixConnector()

    for item in s.list_datasources():

        yield item['name']

from pydantic import BaseModel, Schema

from typing import Dict, List

from datetime import datetime

from fastapi import HTTPException

import logging

from time import time

class LinkProvider(BaseModel):

    name:str = Schema(None, title= "The name of the Provider that provides the links", max_length= 300)

    totalRelationships:int=Schema(None, title= "The number of links that It provides")

class LinkPublisher(BaseModel):

    name:str = Schema(None, title= "The name of the Publisher that provides the links", max_length= 300)

    totalRelationships:int=Schema(None, title= "The number of links that It provides")

class IdentifierType(BaseModel):

    identifier: str = None

    schema_str:str = Schema(None, alias="schema")

class ScholixProviderType(BaseModel):

    name:str

    identifiers:List[IdentifierType] = []

class RelationshipType(BaseModel):

    name:str

    schema_str:str = Schema(None, alias="schema")

    inverseRelationship:str = None

class CreatorType(BaseModel):

    name: str

    identifiers:List[IdentifierType] = []

class ScholixObjectType(BaseModel):

    subtype:str = None

    type:str = None

class ScholixItemType(BaseModel):

    identifiers:List[IdentifierType] = []

    title:str = None

    objectType:str 

    creators:List[CreatorType] = []    

    publisher:List[ScholixProviderType] =[] 

    objectProvider: List[ScholixProviderType] =[]  

class ScholixType(BaseModel):

    linkProvider:List[ScholixProviderType] =[]    

    publicationDate:str =None

    relationship:RelationshipType=None

    source:ScholixItemType=None

    target:ScholixItemType=None

def convert_response(response):

    log = logging.getLogger("scholexplorer")

    for item in response.hits:

        result = item.__dict__['_d_']

        result['linkProvider'] = result.pop('linkprovider')

        if 'creator' in result['source']:

            result['source']['creators']=result['source'].pop('creator')

        result['source']['identifiers']=result['source'].pop('identifier')

        result['source']['objectProvider'] = [s['provider'] for s in result['source'].get('collectedFrom',[])]

        if 'creator' in result['target']:

            result['target']['creators']=result['target'].pop('creator')

        result['target']['identifiers']=result['target'].pop('identifier')

        if 'collectedFrom' in result['target']:

            result['target']['objectProvider'] = [s['provider'] for s in result['target'].get('collectedFrom',[])]

        else:

            result['target']['objectProvider'] = []

        yield  result

from pydantic import BaseModel, Schema

from typing import Dict, List

from datetime import datetime

from fastapi import HTTPException

import logging

from time import time

rels = dict(issupplementto="IsSupplementTo", issupplementedby="IsSupplementedBy", references="References",

            isreferencedby="IsReferencedBy")

pid_resolver = {

    "pdb": "http://www.rcsb.org/pdb/explore/explore.do?structureId=%s",

    "ncbi-n": "http://www.ncbi.nlm.nih.gov/gquery/?term=%s",

    "ncbi": "http://www.ncbi.nlm.nih.gov/gquery/?term=%s",

    "pmid": "http://www.ncbi.nlm.nih.gov/pubmed/%s",

    "pmcid": "http://www.ncbi.nlm.nih.gov/pmc/articles/%s",

    "pubmedid": "http://www.ncbi.nlm.nih.gov/pubmed/%s",

    "doi": "http://dx.doi.org/%s",

    "genbank": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "nuccore": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "swiss-prot": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "arrayexpress": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "biomodels": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "bmrb": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "ena": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "geo": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "ensembl": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "mgi": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "bind": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "pride": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "ddbj": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "bioproject": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "embl": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

    "sra": "http://www.ncbi.nlm.nih.gov/nucest/%s?report=genbank",

}

def resolveIdentifier(pid, pid_type):

    if pid_type != None:

        if pid_type.lower() in pid_resolver:

            return pid_resolver[pid_type.lower()] % pid

        else:

            return "http://identifiers.org/%s:%s" % (pid_type, pid)

class LinkProvider(BaseModel):

    name:str = Schema(None, title= "The name of the Provider that provides the links", max_length= 300)

    totalRelationships:int=Schema(None, title= "The number of links that It provides")

class LinkPublisher(BaseModel):

    name:str = Schema(None, title= "The name of the Publisher that provides the links", max_length= 300)

    totalRelationships:int=Schema(None, title= "The number of links that It provides")

class IdentifierType(BaseModel):

    ID: str = None

    IDScheme:str = None

    IDURL:str= None

class ScholixProviderType(BaseModel):

    name:str

    identifier:List[IdentifierType] = []

class RelationshipType(BaseModel):

    Name:str

    SubType:str

    SubTypeSchema:str = None

class CreatorType(BaseModel):

    Name: str

    Identifier:List[IdentifierType] = []

class ScholixItemType(BaseModel):

    Identifier:List[IdentifierType] = []

    Title:str = None

    Type:str 

    Creator:List[CreatorType] = []

    PublicationDate:str = None

    Publisher:List[ScholixProviderType] =[]    

class ScholixType(BaseModel):

    HarvestDate:str=None

    LicenseURL:str=None

    LinkProvider:List[ScholixProviderType] =[]

    LinkPublicationDate: str = None

    RelationshipType:RelationshipType

    source:ScholixItemType

    target:ScholixItemType

class PageResultType(BaseModel):

    currentPage:int

    totalLinks:int

    totalPages:int

    result:List[ScholixType] = []

def get_scholix_resource(item):

    title = ''

    if 'title' in item:

        title = item.title

    if len(title):

        if title[0] == '"' and title[-1] == '"':

            title = title[1:-1]

    identifier = [dict(ID=x.identifier, IDScheme=x.schema, IDURL=resolveIdentifier(x.identifier, x.schema)) for x in

                  item.identifier]

    identifier.append(dict(ID=item.dnetIdentifier, IDScheme='D-Net Identifier', IDURL='http://scholexplorer.openaire.eu/index.html#/detail/%s'%item.dnetIdentifier))

    creator = []

    if 'creator' in item:

        creator = [dict(Name=x.name) for x in item.creator]

    publicationDate = None

    if 'publicationDate' in item:

        publicationDate = item.publicationDate

    publisher = []

    if 'publisher' in item:

        publisher = [dict(name= x.name) for x in item.publisher]

    c_type = item.objectType

    if item.objectType == 'publication':

        c_type = 'literature'

    resource = dict(Title=title, Identifier=identifier, Creator=creator, PublicationDate= publicationDate, Publisher = publisher, Type= c_type)

    return resource

def convert_response(response):

    now = datetime.now()

    log = logging.getLogger("scholexplorer")

    start = time()

    for item in response.hits:

        current_item = {'LinkPublicationDate': now.strftime("%Y-%m-%d"), 'HarvestDate': now.strftime("%Y-%m-%d"),

                         "LinkProvider": []}

        for linkProvider in item.linkprovider:

            current_item['LinkProvider'].append(ScholixProviderType(name=linkProvider.name,

                                                     identifier=[IdentifierType(ID=x.identifier, IDScheme=x.schema) for x in

                                                                 linkProvider.identifiers]))

        rel_sub_type = rels.get(item.relationship.name.lower(), "IsRelatedTo")

        current_item['RelationshipType'] = dict(Name=rel_sub_type, SubType=item.relationship.name,

                                                SubTypeSchema=item.relationship.schema)

        current_item['source'] = get_scholix_resource(item.source)

        current_item['target'] = get_scholix_resource(item.target)

        yield current_item

    end = time()

    log.debug("response converted in {} ms".format(end-start))

def create_response(response, current_page):

    log = logging.getLogger("scholexplorer")    

    if current_page > 9999:

        raise HTTPException(status_code=400,detail="MAX NUMBER OF PAGE REACHED")

    start = time()

    result = {'totalLinks': response.hits.total, 'currentPage': current_page /100,

              'totalPages': 1 + response.hits.total / 100, 'result': []}

    result['result'] = convert_response(response)

    end = time()

    log.debug("response created in {} ms".format(end-start))

    return result

from fastapi import FastAPI, Query,HTTPException

from pydantic import BaseModel, Schema

from typing import Dict, List

from v2.responseModel import *

from eu.dnetlib.ScholixConnector import ScholixConnector

from datetime import datetime

import time

import logging

subapi_v2 = FastAPI(title="Scholexplorer API 2.0",

    version="2.0.0",

    description="scholexplorer API version 2.0",openapi_prefix="/v2", docs_url="/ui/")

req_fields = ['sourcePid', 'targetPid', 'sourcePublisher', 'targetPublisher', 'linkProvider']

@subapi_v2.get("/LinkProvider", response_model=List[LinkProvider], tags=['LinkProvider : Operation related to the Link Provider'], description="Return a list of link provider and relative number of relations",summary="Get All Link Providers")

def linkProvider(name: str = Query(None, title="name", description="Filter the link provider by a name")):

        s = ScholixConnector()

        return s.list_datasources(name)

@subapi_v2.get("/LinkPublisher/inSource", response_model=List[LinkPublisher], 

            tags=['LinkPublisher : Operation related to the Link Publisher'], 

            description="Return a List of all Publishers that provide source objects in Scholix links and the total number of links where the source object comes from this publisher",

            summary="Get All Publishers that provide source object")

def linkPublisherInSource(name: str = Query(None, title="name", description="Filter the link publisher by a name")):

        s = ScholixConnector()                

        return s.list_publisher('source',name)

@subapi_v2.get("/LinkPublisher/inTarget", response_model=List[LinkPublisher], 

            tags=['LinkPublisher : Operation related to the Link Publisher'], 

            description="Return a List of all Publishers that provide target objects in Scholix links and the total number of links where the target object comes from this publisher",

            summary="Get All Publishers that provide target object")

def linkPublisherInTarget(name: str = Query(None, title="name", description="Filter the link publisher by a name")):

        s = ScholixConnector()

        return s.list_publisher('target',name)

@subapi_v2.get("/Links", response_model=PageResultType, tags=['Links : Operation related to the Scholix Links'], 

response_description="A list of List of scholix Links following the schema <a href='https://github.com/scholix/schema/tree/master/json/v3/schema.json'>https://github.com/scholix/schema/tree/master/json/v3/schema.json</a>",

description="""Return a List of scholix Links, this method <b>require one of the following parameters </b> (<b>sourcePid, targetPid, sourcePublisher,\

        \ targetPublisher, linkProvider</b>) all parameters can be combined""",summary="Get Scholix Links")

def links(

    linkProvider: str = Query(None, title="linkProvider", description="Filter Scholix relationships collected from a LinkProvider"),

    targetPid: str = Query(None, title="targetPid", description="Filter Scholix relationships having a target pid"),

    targetPublisher: str = Query(None, title="targetPublisher", description="Filter Scholix relationships having a target published in a Publisher named targetPublisher"),

    targetPidType: str = Query(None, title="targetPidType", description="Filter Scholix relationships having a target pid type"),

    targetType: str = Query(None, title="targetType", description="Filter Scholix relationships having a target type (literature, dataset, unknown)"),  

    sourceType: str = Query(None, title="sourceType", description="Filter Scholix relationships having a source type (literature, dataset, unknown)"),  

    sourcePid: str = Query(None, title="sourcePid", description="Filter Scholix relationships having a source pid"),

    sourcePublisher: str = Query(None, title="sourcePublisher", description="Filter Scholix relationships having a source published in a Publisher named sourcePublisher"),

    sourcePidType: str = Query(None, title="sourcePidType", description="Filter Scholix relationships having a source pid type"),

    harvestedAfter: str = Query(None, title="harvestedAfter", description="Filter scholix Links having collected after this date"),

        page:int= Query(None, title="page", description="select page of result"),

    ):

        log = logging.getLogger("scholexplorer")

        log.debug("request Link page")

        if not (sourcePid or targetPid or sourcePublisher or targetPublisher or linkProvider):

                raise HTTPException(status_code=400,detail='The method requires one of the following parameters: sourcePid, targetPid, sourcePublisher, targetPublisher, linkProvider')

        s = ScholixConnector()    

        m_page = 0

        if page:

                m_page= page*100

        return create_response(s.links(provider = linkProvider, s_pid=sourcePid, t_pid=targetPid, s_publisher=sourcePublisher,

                                           t_publisher=targetPublisher, s_pid_type=sourcePidType,

                                           t_pid_type=targetPidType, page=m_page, target_Type=targetType, source_Type=sourceType),m_page)    

import logging

import os

def get_index_properties():

    if 'DLI_CONF_PATH' not in os.environ:

        logging.error("ENVIRONEMENT VARIABLE DLI_CONF_PATH DOES NOT EXISTS ")

        raise Exception("ENVIRONEMENT VARIABLE DLI_CONF_PATH DOES NOT EXISTS ")

    if not os.path.exists(os.environ['DLI_CONF_PATH']):

        logging.error("FILE {} DOES NOT EXISTS ".format(os.environ['DLI_CONF_PATH']))

        raise Exception("FILE {} DOES NOT EXISTS ".format(os.environ['DLI_CONF_PATH']))

    with open(os.environ['DLI_CONF_PATH']) as f:

        p = {}

        for line in f:

            if not line.startswith("#"):

                data = line.strip().split("=")

                p[data[0].strip()] = data[1].strip()

        return p

from elasticsearch import *

from elasticsearch_dsl import *

from os import path

import os

from eu.dnetlib.util import get_index_properties

from elasticsearch_dsl.response import Response

import logging

log = logging.getLogger("scholexplorer")

class ScholixConnector(object):    

    __instance = None

    def __new__(cls):

        if ScholixConnector.__instance is None:

            ScholixConnector.__instance = object.__new__(cls)        

            props = get_index_properties()

            index_name = props['api.index']

            index_host = [x for x in props['es_index'].split(',')]    

            #connections.create_connection(hosts=index_host, timeout=1000)    

            ScholixConnector.__instance.connection = Elasticsearch(hosts=index_host, timeout=1000)

            ScholixConnector.__instance.index_host = index_host

            ScholixConnector.__instance.index_name = index_name

        return ScholixConnector.__instance

    def create_pidType_query(self, value, start):

        args = {start + '.identifier.schema': value}

        return Q('nested', path=start + '.identifier', query=Q('bool', must=[Q('match', **args)]))

    def create_pid_query(self, value, start):

        args = {start + '.identifier.identifier': value.lower()}

        return Q('nested', path=start + '.identifier', query=Q('bool', must=[Q('match', **args)]))

    def create_typology_query(self, value, start):

        args = {start + '.objectType': value}

        return Q('nested', path=start, query=Q('bool', must=[Q('match', **args)]))

    def create_dataSource_query(self, value):

        args = {'linkprovider.name': value}

        return Q('nested', path='linkprovider', query=Q('bool', must=[Q('match', **args)]))

    def create_publisher_query(self, value, start):

        args = {start + '.publisher.name': value}

        q = Q('nested', path=start + '.publisher', query=Q('bool', must=[Q('match', **args)]))

        return Q('nested', path=start, query=q)

    def list_datasources(self, ds_name=None):

        search_object = Search(using=self.connection, index=self.index_name).doc_type('scholix')

        if ds_name:

            search_object = search_object.query(self.create_dataSource_query(ds_name))

        else:

            search_object = search_object.query()

        search_object.aggs.bucket('all_datasources', 'nested', path='linkprovider').bucket('all_names', 'terms',

                                                                                           field='linkprovider.name',

                                                                                           size=100)

        response = search_object.execute()

        if ds_name:

            for item in response.aggs.all_datasources.all_names.buckets:

                if item.key == ds_name:

                    yield dict(name=item.key, totalRelationships=item.doc_count)

        else:

            for item in response.aggs.all_datasources.all_names.buckets:

                yield dict(name=item.key, totalRelationships=item.doc_count)    

    def list_publisher(self, start, pub_name=None):

        log.info("Started Index from host")

        search_object = Search(using=self.connection, index=self.index_name).doc_type('scholix')

        if pub_name:

            search_object = search_object.query(self.create_publisher_query(pub_name, start))

        else:

            search_object = search_object.query()

        search_object.aggs.bucket('all_targets', 'nested', path=start).bucket('all_t_pubs', 'nested',

                                                                              path=start + '.publisher').bucket(

            'all_pubs', 'terms',

            field=start + '.publisher.name',

            size=1000)

        response = search_object.execute()        

        for item in response.aggs.all_targets.all_t_pubs.all_pubs.buckets:

            if pub_name and item.key == pub_name:            

                yield dict(name=item.key, totalRelationships=item.doc_count)

            else:

                yield dict(name=item.key, totalRelationships=item.doc_count)

    def links(self, provider=None, s_pid=None, t_pid=None, s_publisher=None, t_publisher=None, s_pid_type=None,

              t_pid_type=None, target_Type=None, source_Type=None,page=0):       

        queries = []

        if provider:

            log.info("PROVIDER NOT NONE: {}".format(provider))

            queries.append(self.create_dataSource_query(provider))

        if s_pid:

            log.info("S_PID NOT NONE: {}".format(s_pid))

            queries.append(self.create_pid_query(s_pid, 'source'))

        if t_pid:

            queries.append(self.create_pid_query(t_pid, 'target'))

        if s_publisher:

            queries.append(self.create_publisher_query(s_publisher, 'source'))

        if t_publisher:

            queries.append(self.create_publisher_query(t_publisher, 'target'))

        if s_pid_type:

            queries.append(self.create_pidType_query(s_pid_type, 'source'))

        if t_pid_type:

            queries.append(self.create_pidType_query(t_pid_type, 'target'))

        if target_Type:

            if 'literature' == target_Type:

                target_Type = 'publication'

            queries.append(self.create_typology_query(target_Type,'target'))

        if source_Type:

            if 'literature' == source_Type:

                source_Type = 'publication'

            queries.append(self.create_typology_query(source_Type,'source'))

        q = None

        for item in queries:

            if not q:

                q = item

            else:

                q = q & item        

        log.debug("REQUEST CREATED {}".format(q))

        search_object = Search(using=self.connection, index=self.index_name).doc_type('scholix').query(q)

        log.debug("Page request size is {}".format(page))

        if page > 9999:

            return []

        return search_object[page:page + 100].execute()

    def realtionToPid(self, pid, pidType=None, datasource=None, typology=None, page=0):

        if pidType:

            query = self.create_pid_pidType_query(pidType.lower(), pid.lower())

        else:

            query = self.create_source_pid_query(pid.lower())

        filters = []

        if datasource and len(datasource):

            filters.append(self.create_dataSource_query(datasource))

        if typology and len(typology):

            filters.append(self.create_typology_query(typology,'target'))

        search_object = Search(using=self.connection, index=self.index_name).doc_type('scholix').query(query)

        if len(filters):

            search_object = search_object.filter(Q('bool', must=filters))

            if page > 9999:

                return []

        return search_object[page:page + 100].execute()

    def realtionToTypology(self, typology, page=0):

        search_object = Search(using=self.connection, index=self.index_name).doc_type('scholix').query(

            self.create_typology_query(typology,'target'))

        if page > 9999:

            return []

        return search_object[page:page + 100].execute()

import time

from prometheus_client import Counter, Gauge, Histogram

from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint

from starlette.requests import Request

from starlette.responses import Response

REQUESTS = Counter("starlette_requests_total", "Total count of requests by method and path.", ["method", "path"])

RESPONSES = Counter(

    "scholexplorerAPI_responses_total",

    "Total count of responses by method, path and status codes.",

    ["method", "path", "status_code"],

)

REQUESTS_PROCESSING_TIME = Histogram(

    "scholexplorerAPI_requests_processing_time_seconds",

    "Histogram of requests processing time by path (in seconds)",

    ["method", "path"],

)

EXCEPTIONS = Counter(

    "scholexplorerAPI_exceptions_total",

    "Histogram of exceptions raised by path and exception type",

    ["method", "path", "exception_type"],

)

REQUESTS_IN_PROGRESS = Gauge(

    "scholexplorerAPI_requests_in_progress",

    "Gauge of requests by method and path currently being processed",

    ["method", "path"],

)

class PrometheusMiddleware(BaseHTTPMiddleware):

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:

        method = request.method

        path = request.url.path

        REQUESTS_IN_PROGRESS.labels(method=method, path=path).inc()

        REQUESTS.labels(method=method, path=path).inc()

        try:

            before_time = time.time()

            response = await call_next(request)

            after_time = time.time()

        except Exception as e:

            EXCEPTIONS.labels(method=method, path=path, exception_type=type(e).__name__).inc()

            raise e from None

        else:

            REQUESTS_PROCESSING_TIME.labels(method=method, path=path).observe(after_time - before_time)

            RESPONSES.labels(method=method, path=path, status_code=response.status_code).inc()

        finally:

            REQUESTS_IN_PROGRESS.labels(method=method, path=path).dec()

        return response

import os

from prometheus_client import CONTENT_TYPE_LATEST, REGISTRY, CollectorRegistry, generate_latest

from prometheus_client.multiprocess import MultiProcessCollector

from starlette.requests import Request

from starlette.responses import Response

def metrics(request: Request) -> Response:

    if "prometheus_multiproc_dir" in os.environ:

        registry = CollectorRegistry()

        MultiProcessCollector(registry)

    else:

        registry = REGISTRY

    return Response(generate_latest(registry), media_type=CONTENT_TYPE_LATEST)

from fastapi import FastAPI

from v2.api_v2 import subapi_v2

from v1.api import subapi_v1

from pympler import muppy, summary

import logging

from eu.dnetlib.metrics_utils import metrics 

from eu.dnetlib.metricsMiddleWare import PrometheusMiddleware

import sys

app = FastAPI()

log = logging.getLogger("scholexplorer")

log.setLevel(logging.INFO)

fh = logging.StreamHandler(sys.stdout) 

fh.setLevel(logging.INFO)

formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

fh.setFormatter(formatter)

log.addHandler(fh)

log.info("Scholexplorer Api Restarted")

app.add_middleware(PrometheusMiddleware)

app.add_route("/metrics/", metrics)

app.mount("/v2", subapi_v2)

app.mount("/v1", subapi_v1)