Spaces:
Running
Running
File size: 2,545 Bytes
75d0d36 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 |
import os
from dotenv import load_dotenv
load_dotenv()
from sqlalchemy import create_engine, text
from fastapi import FastAPI, Depends
DATABASE_URL = os.getenv('DATABASE_URL')
def get_engine():
if not DATABASE_URL:
raise ValueError("L'URL de la base de données (PG_URL) n'est pas définie dans l'environnement.")
engine = create_engine(DATABASE_URL)
return engine
def get_db_connection(engine=Depends(get_engine)):
with engine.connect() as connection:
yield connection
def load_job_offer(connection):
query_offre = text('SELECT entreprise, ville, poste, contrat, description_poste, publication, lien, id FROM "jobs_offers"')
resultats = connection.execute(query_offre).fetchall()
offres = []
for row in resultats:
entreprise, ville, poste, contrat, description_poste, publication, lien, id = row
offres.append({
"entreprise": entreprise,
"ville": ville,
"poste": poste,
"contrat": contrat,
"description_poste": description_poste,
"publication": publication,
"lien": lien,
"id": id
})
return offres
def get_job_offer_by_id(connection, offre_id: str):
query = text('SELECT entreprise, ville, poste, contrat, description_poste, publication, lien, id FROM "jobs_offers" WHERE id = :id')
result = connection.execute(query, {"id": offre_id}).fetchone()
if result:
entreprise, ville, poste, contrat, description_poste, publication, lien, id = result
return {
"entreprise": entreprise,
"ville": ville,
"poste": poste,
"contrat": contrat,
"description_poste": description_poste,
"publication": publication,
"lien": lien,
"id": id
}
return None
app = FastAPI()
@app.get("/offre-emploi")
async def read_job_offer(connection = Depends(get_db_connection)):
offres = load_job_offer(connection)
if offres:
return offres
return {"message": "Aucune offre d'emploi trouvée."}
@app.get("/offre-emploi/{offre_id}")
async def read_job_offer_by_id(offre_id: str, connection = Depends(get_db_connection)):
offre = get_job_offer_by_id(connection, offre_id)
if offre:
return offre
return {"message": f"Aucune offre trouvée avec l'id {offre_id}"}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8010) |