my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	import logging
#2	import os
#3
#4	import aiofiles
#5	import yaml
#6	from database import Base, SessionLocal, engine
#7	from fastapi import Depends, FastAPI, HTTPException, UploadFile
#8	from models import DefaultResponse, DeployAppRequest, QueryApp, SourceApp
#9	from services import get_app, get_apps, remove_app, save_app
#10	from sqlalchemy.orm import Session
#11	from utils import generate_error_message_for_api_keys
#12
#13	from embedchain import App
#14	from embedchain.client import Client
#15
#16	logger = logging.getLogger(__name__)
#17
#18	Base.metadata.create_all(bind=engine)
#19
#20
#21	def get_db():
#22	db = SessionLocal()
#23	try:
#24	yield db
#25	finally:
#26	db.close()
#27
#28
#29	app = FastAPI(
#30	title="Embedchain REST API",
#31	description="This is the REST API for Embedchain.",
#32	version="0.0.1",
#33	license_info={
#34	"name": "Apache 2.0",
#35	"url": "https://github.com/embedchain/embedchain/blob/main/LICENSE",
#36	},
#37	)
#38
#39
#40	@app.get("/ping", tags=["Utility"])
#41	def check_status():
#42	"""
#43	Endpoint to check the status of the API
#44	"""
#45	return {"ping": "pong"}
#46
#47
#48	@app.get("/apps", tags=["Apps"])
#49	async def get_all_apps(db: Session = Depends(get_db)):
#50	"""
#51	Get all apps.
#52	"""
#53	apps = get_apps(db)
#54	return {"results": apps}
#55
#56
#57	@app.post("/create", tags=["Apps"], response_model=DefaultResponse)
#58	async def create_app_using_default_config(app_id: str, config: UploadFile = None, db: Session = Depends(get_db)):
#59	"""
#60	Create a new app using App ID.
#61	If you don't provide a config file, Embedchain will use the default config file\n
#62	which uses opensource GPT4ALL model.\n
#63	app_id: The ID of the app.\n
#64	config: The YAML config file to create an App.\n
#65	"""
#66	try:
#67	if app_id is None:
#68	raise HTTPException(detail="App ID not provided.", status_code=400)
#69
#70	if get_app(db, app_id) is not None:
#71	raise HTTPException(detail=f"App with id '{app_id}' already exists.", status_code=400)
#72
#73	yaml_path = "default.yaml"
#74	if config is not None:
#75	contents = await config.read()
#76	try:
#77	yaml.safe_load(contents)
#78	# TODO: validate the config yaml file here
#79	yaml_path = f"configs/{app_id}.yaml"
#80	async with aiofiles.open(yaml_path, mode="w") as file_out:
#81	await file_out.write(str(contents, "utf-8"))
#82	except yaml.YAMLError as exc:
#83	raise HTTPException(detail=f"Error parsing YAML: {exc}", status_code=400)
#84
#85	save_app(db, app_id, yaml_path)
#86
#87	return DefaultResponse(response=f"App created successfully. App ID: {app_id}")
#88	except Exception as e:
#89	logger.warning(str(e))
#90	raise HTTPException(detail=f"Error creating app: {str(e)}", status_code=400)
#91
#92
#93	@app.get(
#94	"/{app_id}/data",
#95	tags=["Apps"],
#96	)
#97	async def get_datasources_associated_with_app_id(app_id: str, db: Session = Depends(get_db)):
#98	"""
#99	Get all data sources for an app.\n
#100	app_id: The ID of the app. Use "default" for the default app.\n
#101	"""
#102	try:
#103	if app_id is None:
#104	raise HTTPException(
#105	detail="App ID not provided. If you want to use the default app, use 'default' as the app_id.",
#106	status_code=400,
#107	)
#108
#109	db_app = get_app(db, app_id)
#110
#111	if db_app is None:
#112	raise HTTPException(detail=f"App with id {app_id} does not exist, please create it first.", status_code=400)
#113
#114	app = App.from_config(config_path=db_app.config)
#115
#116	response = app.get_data_sources()
#117	return {"results": response}
#118	except ValueError as ve:
#119	logger.warning(str(ve))
#120	raise HTTPException(
#121	detail=generate_error_message_for_api_keys(ve),
#122	status_code=400,
#123	)
#124	except Exception as e:
#125	logger.warning(str(e))
#126	raise HTTPException(detail=f"Error occurred: {str(e)}", status_code=400)
#127
#128
#129	@app.post(
#130	"/{app_id}/add",
#131	tags=["Apps"],
#132	response_model=DefaultResponse,
#133	)
#134	async def add_datasource_to_an_app(body: SourceApp, app_id: str, db: Session = Depends(get_db)):
#135	"""
#136	Add a source to an existing app.\n
#137	app_id: The ID of the app. Use "default" for the default app.\n
#138	source: The source to add.\n
#139	data_type: The data type of the source. Remove it if you want Embedchain to detect it automatically.\n
#140	"""
#141	try:
#142	if app_id is None:
#143	raise HTTPException(
#144	detail="App ID not provided. If you want to use the default app, use 'default' as the app_id.",
#145	status_code=400,
#146	)
#147
#148	db_app = get_app(db, app_id)
#149
#150	if db_app is None:
#151	raise HTTPException(detail=f"App with id {app_id} does not exist, please create it first.", status_code=400)
#152
#153	app = App.from_config(config_path=db_app.config)
#154
#155	response = app.add(source=body.source, data_type=body.data_type)
#156	return DefaultResponse(response=response)
#157	except ValueError as ve:
#158	logger.warning(str(ve))
#159	raise HTTPException(
#160	detail=generate_error_message_for_api_keys(ve),
#161	status_code=400,
#162	)
#163	except Exception as e:
#164	logger.warning(str(e))
#165	raise HTTPException(detail=f"Error occurred: {str(e)}", status_code=400)
#166
#167
#168	@app.post(
#169	"/{app_id}/query",
#170	tags=["Apps"],
#171	response_model=DefaultResponse,
#172	)
#173	async def query_an_app(body: QueryApp, app_id: str, db: Session = Depends(get_db)):
#174	"""
#175	Query an existing app.\n
#176	app_id: The ID of the app. Use "default" for the default app.\n
#177	query: The query that you want to ask the App.\n
#178	"""
#179	try:
#180	if app_id is None:
#181	raise HTTPException(
#182	detail="App ID not provided. If you want to use the default app, use 'default' as the app_id.",
#183	status_code=400,
#184	)
#185
#186	db_app = get_app(db, app_id)
#187
#188	if db_app is None:
#189	raise HTTPException(detail=f"App with id {app_id} does not exist, please create it first.", status_code=400)
#190
#191	app = App.from_config(config_path=db_app.config)
#192
#193	response = app.query(body.query)
#194	return DefaultResponse(response=response)
#195	except ValueError as ve:
#196	logger.warning(str(ve))
#197	raise HTTPException(
#198	detail=generate_error_message_for_api_keys(ve),
#199	status_code=400,
#200	)
#201	except Exception as e:
#202	logger.warning(str(e))
#203	raise HTTPException(detail=f"Error occurred: {str(e)}", status_code=400)
#204
#205
#206	# FIXME: The chat implementation of Embedchain needs to be modified to work with the REST API.
#207	# @app.post(
#208	# "/{app_id}/chat",
#209	# tags=["Apps"],
#210	# response_model=DefaultResponse,
#211	# )
#212	# async def chat_with_an_app(body: MessageApp, app_id: str, db: Session = Depends(get_db)):
#213	# """
#214	# Query an existing app.\n
#215	# app_id: The ID of the app. Use "default" for the default app.\n
#216	# message: The message that you want to send to the App.\n
#217	# """
#218	# try:
#219	# if app_id is None:
#220	# raise HTTPException(
#221	# detail="App ID not provided. If you want to use the default app, use 'default' as the app_id.",
#222	# status_code=400,
#223	# )
#224
#225	# db_app = get_app(db, app_id)
#226
#227	# if db_app is None:
#228	# raise HTTPException(
#229	# detail=f"App with id {app_id} does not exist, please create it first.",
#230	# status_code=400
#231	# )
#232
#233	# app = App.from_config(config_path=db_app.config)
#234
#235	# response = app.chat(body.message)
#236	# return DefaultResponse(response=response)
#237	# except ValueError as ve:
#238	# raise HTTPException(
#239	# detail=generate_error_message_for_api_keys(ve),
#240	# status_code=400,
#241	# )
#242	# except Exception as e:
#243	# raise HTTPException(detail=f"Error occurred: {str(e)}", status_code=400)
#244
#245
#246	@app.post(
#247	"/{app_id}/deploy",
#248	tags=["Apps"],
#249	response_model=DefaultResponse,
#250	)
#251	async def deploy_app(body: DeployAppRequest, app_id: str, db: Session = Depends(get_db)):
#252	"""
#253	Query an existing app.\n
#254	app_id: The ID of the app. Use "default" for the default app.\n
#255	api_key: The API key to use for deployment. If not provided,
#256	Embedchain will use the API key previously used (if any).\n
#257	"""
#258	try:
#259	if app_id is None:
#260	raise HTTPException(
#261	detail="App ID not provided. If you want to use the default app, use 'default' as the app_id.",
#262	status_code=400,
#263	)
#264
#265	db_app = get_app(db, app_id)
#266
#267	if db_app is None:
#268	raise HTTPException(detail=f"App with id {app_id} does not exist, please create it first.", status_code=400)
#269
#270	app = App.from_config(config_path=db_app.config)
#271
#272	api_key = body.api_key
#273	# this will save the api key in the embedchain.db
#274	Client(api_key=api_key)
#275
#276	app.deploy()
#277	return DefaultResponse(response="App deployed successfully.")
#278	except ValueError as ve:
#279	logger.warning(str(ve))
#280	raise HTTPException(
#281	detail=generate_error_message_for_api_keys(ve),
#282	status_code=400,
#283	)
#284	except Exception as e:
#285	logger.warning(str(e))
#286	raise HTTPException(detail=f"Error occurred: {str(e)}", status_code=400)
#287
#288
#289	@app.delete(
#290	"/{app_id}/delete",
#291	tags=["Apps"],
#292	response_model=DefaultResponse,
#293	)
#294	async def delete_app(app_id: str, db: Session = Depends(get_db)):
#295	"""
#296	Delete an existing app.\n
#297	app_id: The ID of the app to be deleted.
#298	"""
#299	try:
#300	if app_id is None:
#301	raise HTTPException(
#302	detail="App ID not provided. If you want to use the default app, use 'default' as the app_id.",
#303	status_code=400,
#304	)
#305
#306	db_app = get_app(db, app_id)
#307
#308	if db_app is None:
#309	raise HTTPException(detail=f"App with id {app_id} does not exist, please create it first.", status_code=400)
#310
#311	app = App.from_config(config_path=db_app.config)
#312
#313	# reset app.db
#314	app.db.reset()
#315
#316	remove_app(db, app_id)
#317	return DefaultResponse(response=f"App with id {app_id} deleted successfully.")
#318	except Exception as e:
#319	raise HTTPException(detail=f"Error occurred: {str(e)}", status_code=400)
#320
#321
#322	if __name__ == "__main__":
#323	import uvicorn
#324
#325	is_dev = os.getenv("DEVELOPMENT", "False")
#326	uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=bool(is_dev))
#327

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public