from fastapi import FastAPI, Depends, HTTPException, Request, Form, status from fastapi.responses import RedirectResponse from fastapi.templating import Jinja2Templates from sqlalchemy import create_engine, Column, Integer, String, Date, Identity, Boolean, Text , TIMESTAMP from sqlalchemy.orm import sessionmaker, declarative_base, Session from geoalchemy2 import Geometry, Geography #from keycloak import KeycloakOpenID from pydantic import BaseModel from datetime import date, datetime # Configurazione Keycloak # keycloak_openid = KeycloakOpenID(server_url=os.getenv('KEYCLOAK_URL'), # client_id=os.getenv('KEYCLOAK_CLIENT_ID'), # realm_name=os.getenv('KEYCLOAK_REALM'), # client_secret_key=os.getenv('KEYCLOAK_CLIENT_SECRET')) # Template Jinja2 templates = Jinja2Templates(directory="templates") # Configurazione del database PostgreSQL SQLALCHEMY_DATABASE_URL = "postgresql://postgres:postgres@165.22.75.145:15432/GenerationDAITA25" #SQLALCHEMY_DATABASE_URL = "postgresql://postgres:postgres@165.22.75.145:15432/backend" engine = create_engine(SQLALCHEMY_DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) def get_db(): db = SessionLocal() try: yield db finally: db.close() # FastAPI app = FastAPI() Base = declarative_base() class TabellaEdifici(Base): __tablename__ = 'edifici' # codice_catastale della tabella nel database id_ = Column(Integer, primary_key=True, name='id_') # Identity() per autoincrement in PostgreSQL id_codice_fiscale = Column(String(16), name='id_codice_fiscale') id_edificio_osm = Column(String(30), nullable=False, name='id_edificio') indirizzo = Column(String(100), nullable=False) codice_catastale = Column(String(50), nullable=True) coordinate = Column(Geometry(geometry_type='POINT', srid=4326)) # SRID 4326 è lo standard per latitudine/longitudine poligono = Column(Geometry(geometry_type='POLYGON', srid=4326)) # Poligono tipo_edificio = Column(String(100)) stato = Column(Boolean, default=True) data_installazione = Column(Date, nullable=False) data_cancellazione = Column(Date, nullable=True) class TabellaTFO(Base): # Puoi scegliere un nome più descrittivo per la tua tabella __tablename__ = 'tfo' # Scegli un nome per la tabella nel database id_ = Column(Integer, primary_key=True, name='id_') # Chiave primaria, autoincrement id_codice_fiscale = Column(String(16), name='id_codice_fiscale') id_tfo = Column(String(30), nullable=False, name='id_tfo') # Not Null codice_catastale = Column(String(50), nullable=True, name='codice_catastale') # Not Null operatore = Column(String(255)) piano = Column(String(50)) scala = Column(String(50)) interno = Column(String(50)) id_edificio = Column(String(30), nullable=False, name='id_edificio') # Not Null data_installazione = Column(Date, nullable=False, name='data_installazione') data_cancellazione = Column(Date, nullable=True, name='data_cancellazione') class TabellaUtenti(Base): # Nome descrittivo per la tabella __tablename__ = 'utenti' # Nome tabella nel database id_ = Column(Integer, primary_key=True, name='id_') # Chiave primaria, autoincrement codice_fiscale = Column(String(16), name='codice_fiscale') nome = Column(String(255), nullable=False) # Not Null cognome = Column(String(255), nullable=False) # Not Null email = Column(String(100), nullable=False) # Not Null ente = Column(String(255)) ruolo = Column(String(255)) stato_utente = Column(Boolean) class TabellaLogEventi(Base): # Nome descrittivo per la tabella di log __tablename__ = 'log_' # Nome tabella nel database (scegli un nome appropriato) id_ = Column(Integer, primary_key=True, name='id_') # Chiave primaria, autoincrement id_edificio = Column(Integer, name='id_edificio') id_tfo = Column(String(30), name='id_tfo') id_utente = Column(String(16), name='id_utente') data = Column(TIMESTAMP, default =datetime.now(), name='data') # Timestamp senza timezone, Not Null, default CURRENT_TIMESTAMP categoria_modifica = Column(String(255), name='categoria_modifica') descrizione_evento = Column(String(4000), name='descrizione_evento') class Assistenza(Base): __tablename__ = 'assistenza' id_ticket = Column(Integer, primary_key=True, autoincrement=True) nome = Column(String(50)) cognome = Column(String(50)) email = Column(String(100), unique=True, nullable=False) data_richiesta = Column(TIMESTAMP, default=func.current_timestamp()) descrizione_problema = Column(String(4000)) # Base.metadata.drop_all(engine) # Base.metadata.create_all(bind=engine) #### accesso_modifica = "Francesco" #### def get_buildings_by_user(db): return db.query(TabellaEdifici).filter(TabellaEdifici.id_codice_fiscale == accesso_modifica).all() def get_specific_building(db, building_id): return db.query(TabellaEdifici).filter(TabellaEdifici.id_edificio_osm == building_id).first() ############Funzione per filtro ricerca def filtro_ricerca_indirizzo(db, filtro): return db.query(TabellaEdifici).filter(TabellaEdifici.indirizzo.ilike(f"%{filtro}%")).all() def filtro_ricerca_id_edificio_osm(db, filtro): return db.query(TabellaEdifici).filter(TabellaEdifici.id_edificio_osm.ilike(f"%{filtro}%")).all() def filtro_ricerca_data(db, filtro): return db.query(TabellaEdifici).filter(TabellaEdifici.data_installazione == filtro).all() def filtro_ricerca_tipo_edificio(db, filtro): return db.query(TabellaEdifici).filter(TabellaEdifici.tipo_edificio.ilike(f"%{filtro}%")).all() def filtro_ricerca_id_codice_fiscale(db, filtro): return db.query(TabellaEdifici).filter(TabellaEdifici.id_codice_fiscale.ilike(f"%{filtro}%")).all() def filtro_ricerca_totale(db, filtro_indirizzo, filtro_data, filtro_id_edificio_osm, filtro_tipo_edificio, filtro_id_codice_fiscale): query1 = db.query(TabellaEdifici) if filtro_indirizzo is not None: query1 = query1.filter(TabellaEdifici.indirizzo.ilike(f"%{filtro_indirizzo}%")) if filtro_data is not None: query1 = query1.filter(TabellaEdifici.data_installazione.ilike(f"%{filtro_data}%")) if filtro_id_edificio_osm is not None: query1 = query1.filter(TabellaEdifici.id_edificio_osm.ilike(f"%{filtro_id_edificio_osm}%")) if filtro_tipo_edificio is not None: query1 = query1.filter(TabellaEdifici.tipo_edificio.ilike(f"%{filtro_tipo_edificio}%")) if filtro_id_codice_fiscale is not None: query1 = query1.filter(TabellaEdifici.id_codice_fiscale.ilike(f"%{filtro_id_codice_fiscale}%")) return query1.all() @app.get("/") async def home(request: Request): return templates.TemplateResponse("login.html", {"request": request}) # @app.post("/login") # async def login(request: Request, code: str = Form(...)): # token = keycloak_openid.token(grant_type='authorization_code', code=code, redirect_uri=os.getenv('REDIRECT_URI')) # if not token: # raise HTTPException(status_code=401, detail="Login fallito") # return RedirectResponse(url="/inserisci_dati") # @app.get("/inserisci_dati") # async def inserisci_dati(request: Request): # return templates.TemplateResponse("form.html", {"request": request}) # @app.post("/salva_dati") # async def salva_dati(request: Request, # id_codice_fiscale: str = Form(...), # id_edificio_osm: str = Form(...), # indirizzo: str = Form(...), # codice_catastale: str = Form(...), # coordinate: Geometry = Form(...), # poligono: Geometry = Form(...), # tipo_edificio: str = Form(...), # data_installazione: Geometry = Form(...), # stato: int = Form(...), # db: Session = Depends(get_db)): # edificio = TabellaEdifici(id_codice_fiscale=id_codice_fiscale, # id_edificio_osm=id_edificio_osm, # indirizzo=indirizzo, # codice_catastale=codice_catastale, # coordinate=coordinate, # poligono=poligono, # tipo_edificio=tipo_edificio, # data_installazione=data_installazione, # stato=stato) # db.add(edificio) # db.commit() # db.refresh(edificio) # return RedirectResponse(url="/salva_dati", status_code=status.HTTP_303_SEE_OTHER) @app.get("/visualizza_pratiche") async def visualizza_pratiche(request: Request, db: Session = Depends(get_db)): return get_buildings_by_user(db) @app.post("/visualizza_pratiche/modifica") async def modifica(request: Request, building_id : str = Form(...), indirizzo: str = Form(...), codice_catastale: str = Form(...), tipo_edificio: str = Form(...), db: Session = Depends(get_db)): pratica_da_modificare = get_specific_building(db, building_id) if pratica_da_modificare.indirizzo != indirizzo: pratica_da_modificare.indirizzo = indirizzo if pratica_da_modificare.codice_catastale != codice_catastale: pratica_da_modificare.codice_catastale = codice_catastale if pratica_da_modificare.tipo_edificio != tipo_edificio: pratica_da_modificare.tipo_edificio = tipo_edificio db.commit() return RedirectResponse(url="/visualizza_pratiche", status_code=status.HTTP_303_SEE_OTHER) @app.post("/visualizza_pratiche/elimina") async def elimina(request: Request, building_id : str = Form(...), db: Session = Depends(get_db)): pratica_da_eliminare = get_specific_building(db, building_id) pratica_da_eliminare.data_cancellazione = date.today() db.commit() return RedirectResponse(url="/visualizza_pratiche", status_code=status.HTTP_303_SEE_OTHER) @app.post("/visualizza_pratiche/ripristina") async def ripristina(request: Request, building_id : str = Form(...), db: Session = Depends(get_db)): pratica_da_ripristinare = get_specific_building(db, building_id) pratica_da_ripristinare.data_cancellazione = None db.commit() return RedirectResponse(url="/visualizza_pratiche", status_code=status.HTTP_303_SEE_OTHER) def aggiungi_prova(): with SessionLocal() as db: edificio = TabellaEdifici(id_codice_fiscale="RSSMRA80A01H501Z", id_edificio_osm="ED006", indirizzo="corso italia 1", codice_catastale="1244", coordinate="POINT(12.496365 41.902782)", poligono="POLYGON((12.496365 41.902782, 12.496365 41.902782, 12.496365 41.902782, 12.496365 41.902782, 12.496365 41.902782))", tipo_edificio="commerciale", data_installazione=datetime.strptime("12/11/2021", "%d/%m/%Y").date() ) db.add(edificio) db.commit() aggiungi_prova() # uvicorn main:app --reload