Spaces:
Runtime error
Runtime error
# coding=utf-8 | |
# Copyright 2021-present, the Recognai S.L. team. | |
# | |
# Licensed under the Apache License, Version 2.0 (the "License"); | |
# you may not use this file except in compliance with the License. | |
# You may obtain a copy of the License at | |
# | |
# http://www.apache.org/licenses/LICENSE-2.0 | |
# | |
# Unless required by applicable law or agreed to in writing, software | |
# distributed under the License is distributed on an "AS IS" BASIS, | |
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
# See the License for the specific language governing permissions and | |
# limitations under the License. | |
""" | |
This module configures the global fastapi application | |
""" | |
import inspect | |
import os | |
import sys | |
import warnings | |
from pathlib import Path | |
from brotli_asgi import BrotliMiddleware | |
from fastapi import FastAPI | |
from fastapi.exceptions import RequestValidationError | |
from fastapi.middleware.cors import CORSMiddleware | |
from pydantic import ConfigError | |
from argilla import __version__ as argilla_version | |
from argilla.logging import configure_logging | |
from argilla.server.daos.backend.elasticsearch import ( | |
ElasticsearchBackend, | |
GenericSearchError, | |
) | |
from argilla.server.daos.datasets import DatasetsDAO | |
from argilla.server.daos.records import DatasetRecordsDAO | |
from argilla.server.errors import APIErrorHandler, EntityNotFoundError | |
from argilla.server.routes import api_router | |
from argilla.server.security import auth | |
from argilla.server.settings import settings | |
from argilla.server.static_rewrite import RewriteStaticFiles | |
def configure_middleware(app: FastAPI): | |
"""Configures fastapi middleware""" | |
app.add_middleware( | |
CORSMiddleware, | |
allow_origins=settings.cors_origins, | |
allow_credentials=True, | |
allow_methods=["*"], | |
allow_headers=["*"], | |
) | |
app.add_middleware(BrotliMiddleware, minimum_size=512, quality=7) | |
def configure_api_exceptions(api: FastAPI): | |
"""Configures fastapi exception handlers""" | |
api.exception_handler(EntityNotFoundError)(APIErrorHandler.common_exception_handler) | |
api.exception_handler(Exception)(APIErrorHandler.common_exception_handler) | |
api.exception_handler(RequestValidationError)( | |
APIErrorHandler.common_exception_handler | |
) | |
def configure_api_router(app: FastAPI): | |
"""Configures and set the api router to app""" | |
app.include_router(api_router, prefix="/api") | |
def configure_app_statics(app: FastAPI): | |
"""Configure static folder for app""" | |
parent_path = Path(__file__).parent.absolute() | |
statics_folder = Path(os.path.join(parent_path, "static")) | |
if not (statics_folder.exists() and statics_folder.is_dir()): | |
return | |
app.mount( | |
"/", | |
RewriteStaticFiles( | |
directory=statics_folder, | |
html=True, | |
check_dir=False, | |
), | |
name="static", | |
) | |
def configure_app_storage(app: FastAPI): | |
async def configure_elasticsearch(): | |
try: | |
es_wrapper = ElasticsearchBackend.get_instance() | |
dataset_records: DatasetRecordsDAO = DatasetRecordsDAO(es_wrapper) | |
datasets: DatasetsDAO = DatasetsDAO.get_instance( | |
es_wrapper, records_dao=dataset_records | |
) | |
datasets.init() | |
dataset_records.init() | |
except GenericSearchError as error: | |
raise ConfigError( | |
f"Your Elasticsearch endpoint at {settings.obfuscated_elasticsearch()} " | |
"is not available or not responding.\n" | |
"Please make sure your Elasticsearch instance is launched and correctly running and\n" | |
"you have the necessary access permissions. " | |
"Once you have verified this, restart the argilla server.\n" | |
) from error | |
def configure_app_security(app: FastAPI): | |
if hasattr(auth, "router"): | |
app.include_router(auth.router) | |
def configure_app_logging(app: FastAPI): | |
"""Configure app logging using""" | |
app.on_event("startup")(configure_logging) | |
app = FastAPI( | |
title="argilla", | |
description="argilla API", | |
# Disable default openapi configuration | |
openapi_url="/api/docs/spec.json", | |
docs_url="/api/docs" if settings.docs_enabled else None, | |
redoc_url=None, | |
version=str(argilla_version), | |
) | |
def configure_telemetry(app): | |
message = "\n" | |
message += inspect.cleandoc( | |
""" | |
Argilla uses telemetry to report anonymous usage and error information. | |
You can know more about what information is reported at: | |
https://docs.argilla.io/en/latest/reference/telemetry.html | |
Telemetry is currently enabled. If you want to disable it, you can configure | |
the environment variable before relaunching the server: | |
""" | |
) | |
message += "\n\n " | |
message += ( | |
"#set ARGILLA_ENABLE_TELEMETRY=0" | |
if os.name == "nt" | |
else "$>export ARGILLA_ENABLE_TELEMETRY=0" | |
) | |
message += "\n" | |
async def check_telemetry(): | |
if settings.enable_telemetry: | |
print(message, flush=True) | |
for app_configure in [ | |
configure_app_logging, | |
configure_middleware, | |
configure_api_exceptions, | |
configure_app_security, | |
configure_api_router, | |
configure_app_statics, | |
configure_app_storage, | |
configure_telemetry, | |
]: | |
app_configure(app) | |