jueves, 21 de agosto de 2025

WEBPROPv2 (II). Configurar el objeto app de FastHTML (mnu_fh.py). Exception handllers. Beforeware. Mounting folders

0. Introducción

Se ha visto que el módulo que  tiene el "main" es mnu_main.py que ejecuta el servidor web uvicorn al que se el pasa entre otros parámetros el objeto "app".

Pues en este post veremos como se configura el objeto app para que utilice recursos como js o css, utilice rutas estáticas, redirija excepciones y utilice una función para que antes de ir a una ruta o "endpoint" compruebe que el usuario tiene permiso para ello.

1. Configuración del objeto app

Se realiza en el módulo menu_fh.py :

  • Crea la lista de "headers" (líneas 21-43):
    • Utiliza la codificación UTF-8
    • Par un diseño "responsive" utiliza la etiqueta "meta" para asignar toda la anchura del dispositivo y hacer una escalado del 100%
    • Adjunta o enlaza librerias js o css:
      • Font Awesome
      • El css propio para el tree.menu
      • Libreria bootstrap
      • Librería Ag-Grid
  • Crea un manejador de las excepciones 400, 403 y 404 (líneas 45-49)
  • Se define una funcion tipo "Beforeware" que se ejecutará en todos los "endpoints". Se especifica su funcionamiento más adelante (Líneas 52-88)
  • Se montan las carpetas "static" y "downloads" a un path de la app (líneas 93-94)
  • Se añaden estos objetos al objeto app (líneas 86-90)
 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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
#1. Imports

from fastapi.staticfiles import StaticFiles
from fasthtml import common as fh


#------Imprescindible para poder importar de otras carpetas (de basicutils)
import sys
from pathlib import Path
path_root1 = Path(__file__).parents[1] # Damos un salto de directorio hacia arriba-> ximoutilsmod
sys.path.append(str(path_root1))
path_root0 = Path(__file__).parents[0] # El mismo directorio
sys.path.append(str(path_root0))
# ------Fin imprescindible

from basicutils import xmother
from session import xmsession


#1. Defining Headers (Styles, Scripts, etc)
my_headers = [
	fh.Meta(charset='UTF-8'),

	# Responsive design:  Uses the width of the device and renders the page at 100% zoom (no auto zoom)
	fh.Meta(name='viewport', content='width=device-width, initial-scale=1'),
	
	# Link libraries of js and css

	#  1. Font Awesome
	fh.Link(rel='stylesheet', href='https://cdnjs.cloudflare.com/ajax/libs/font-awesome/6.6.0/css/all.min.css'),
	
	#  2. My Menu-tree library
	fh.Link(rel='stylesheet', href='../static/css/tree-menu.css'),
	
	#  3. Bootstrap
	fh.Link(href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/css/bootstrap.min.css", rel="stylesheet"),
        fh.Script(src="https://cdn.jsdelivr.net/npm/@popperjs/core@2.11.8/dist/umd/popper.min.js"),
        fh.Script(src="https://cdn.jsdelivr.net/npm/bootstrap@5.3.3/dist/js/bootstrap.min.js"),
	
	#  4. Ag-Grid
	fh.Script(src="https://cdn.jsdelivr.net/npm/ag-grid-community/dist/ag-grid-community.min.js"),
	
]

exception_handlers={
        400: lambda req, exc: fh.Titled("400: Bad Request: La petició no és correcta."),
	403: lambda req, exc: fh.Titled("403: Forbidden: No té permís a esta pàgina."),
	404: lambda req, exc: fh.Titled("404: Not Found: No existeix esta pàgina!"),
}

#2. Beforeware function to be executed in all endpoints
async def my_before_function(req, tab_id, sess):
	'''
	Beforeware function: checks that the user is authenticated
	'''
	print('Des de beforeware, la URL es:', req.url)
	# 1. Validem que haja tab_id i que l'usuari estiga en la sessió de la BD
	if tab_id is None: tab_id= req.query_params.get('tab_id', None)

	if tab_id is None:	
		try:
			tab_id= (await req.form()).get('tab_id', None)
		except Exception as e:
			print("Error al llegir el tab_id del form:", e)
			tab_id = None
	if tab_id is None:	
		try:
			tab_id= (await req.json()).get('tab_id', None)
		except Exception as e:
			print("Error al llegir el tab_id del JSON:", e)
			tab_id = None	

	my_response,myerror=xmsession.validate(tab_id=tab_id, req=req)
	if (len(myerror.strip())>1):
		print ("redirecting from my_before_function " )
		return my_response

#OJO: Aquelles rutes que comecen per "/softprop/nchk/" no seran comprovades l'usuari
my_skip=[r'/favicon\.ico', r'/static/.*', r'.*\.css', 
		 r'/softprop/nchk/.*',
		 '/softprop/login', '/softprop/validate_login', 
		]
my_beforeware= fh.Beforeware(my_before_function, skip=my_skip)

#3. Creating FastAPI App
app = fh.FastHTML(
	hdrs=my_headers, 
	before=my_beforeware,
	secret_key=xmother.new_id(25), 
	exception_handlers=exception_handlers)

#3. Mounting the static folder for accessing the menu.css
app.mount("/static"   , fh.StaticFiles(directory="static"   ), name="static"   )
app.mount("/downloads", fh.StaticFiles(directory="downloads"), name="downloads")

2. Exception handlers

Es un parámetro del objeto app que és una lista de diccionarios que se le indica el código hhtp del error y lo que se tiene que mostrar. En este caso se muestra una pantalla cuyo título es la descripción del error.

El código se muestra muestra el detalle:

exception_handlers={
        400: lambda req, exc: fh.Titled("400: Bad Request: La petició no és correcta."),
	403: lambda req, exc: fh.Titled("403: Forbidden: No té permís a esta pàgina."),
	404: lambda req, exc: fh.Titled("404: Not Found: No existeix esta pàgina!"),
}


3. La función "beforeware"

Un beforeware es una función que intercepta las peticiones HTTP antes de que lleguen al manejador (es decir, antes del código que atiende /ruta/...), para:
  • Validar autenticación/autorización.
  • Añadir datos a la petición.
  • Redirigir, denegar o registrar información.
  • Omitirse en rutas específicas si se configura.
En este caso, el beforeware comprueba si el usuario está autenticado mediante un tab_id.

Veamos que es lo que contiene el código:
  • Definimos una función asíncrona que recibe como parámetros la "request", el "tab_id" y la "sessión" (Línea 2)
  • Esta función recibe como parámetro el tab_id, y si no se le pasa dicho parámetro lo recoge de la "request" (Línea 8)
  • Si no lo encuentra lo busca primeramente en el FORM i si no en el BODY (Líneas 11-24)
  • Se valida mediante el tab_id con la llamada a la función xmsession.validate () (Línea 26)
  • Se definen aquellas rutas (endpoints) que no hace falta validar con el tab_id, que constituyen una lista de "strings" que puden incluir expresiones regulares. (Líneas 32-35)
  • Se asocia la función anterior como "beforeware global" que actuará en todos los endpoints excepto aquellos que se le indican en el parámetro "skip" (Línea 36)
  • Se le passa el beforeware con el parámetro "before" al objeto "app" (Línea 41)
 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
#2. Beforeware function to be executed in all endpoints
async def my_before_function(req, tab_id, sess):
	'''
	Beforeware function: checks that the user is authenticated
	'''
	print('Des de beforeware, la URL es:', req.url)
	# 1. Validem que haja tab_id i que l'usuari estiga en la sessió de la BD
	if tab_id is None: tab_id= req.query_params.get('tab_id', None)

	# 2. If tab_id is not in the query parameters, we try to get it from the FORM 
	if tab_id is None:	
		try:
			tab_id= (await req.form()).get('tab_id', None)
		except Exception as e:
			print("Error al llegir el tab_id del form:", e)
			tab_id = None
	
	# 3. If tab_id is still None, we try to get it from the JSON body
	if tab_id is None:	
		try:
			tab_id= (await req.json()).get('tab_id', None)
		except Exception as e:
			print("Error al llegir el tab_id del JSON:", e)
			tab_id = None	

	my_response,myerror=xmsession.validate(tab_id=tab_id, req=req)
	if (len(myerror.strip())>1):
		print ("redirecting from my_before_function " )
		return my_response

#OJO: Aquelles rutes que comecen per "/softprop/nchk/" no seran comprovades l'usuari
my_skip=[r'/favicon\.ico', r'/static/.*', r'.*\.css', 
		 r'/softprop/nchk/.*',
		 '/softprop/login', '/softprop/validate_login', 
	]
my_beforeware= fh.Beforeware(my_before_function, skip=my_skip)

#3. Creating FastAPI App
app = fh.FastHTML(
	hdrs=my_headers, 
	before=my_beforeware,
	secret_key=xmother.new_id(25), 
	exception_handlers=exception_handlers)

4. La función xmsession.validate para ver si el usuario tiene permiso

En el módulo xmsession se tiene esta función que :
  • Se crea una ruta de redirección para el caso de error que es "/softprop/login" para que ingrese las credenciales al sistema.
  • El primer error que lnaza es no pasar-le un tab_id
  • Si el tab_id no está presente en las cookies que duran 3600 segundos lanza error
  • Compueba que el tab_id esté en la sessión guardada en la BD 
  • Si hay error lo devuelve a la pantalla de login


 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
def validate(tab_id:str, req)->tuple:
	redirection='/softprop/login'
	my_error=""

	# 1. Si es la primera vegada, no tenim tab_id i envia un script per a arreplegar-lo
	if tab_id is None: 
		my_error="No s'ha inclòs el tab_id en el request"
			
	#2. Check if the tab_id is in the cookies
	#   Cookies that has finished the max_age=3600 seconds are not present in the session
	if len(my_error)==0:
		cookies = req.cookies
		if tab_id not in cookies.keys(): 
			my_error="El tab_id no està present en el cookies, pot estar caducat"

	#3. Verifiquem que l'usuari existisca i estiga actiu
	if len(my_error)==0:
		with xmdb.session_local() as db:
			user, my_error=test_session(req, db, tab_id)
	
			if len(user.strip())==0 and len(my_error.strip())==0:
				my_error="Usuari no trobat"
			
			# 4. Guardem la petició de l'usuari per a fer una auditoria
			else:
				user_request=RequestStore(
					description=str(req.url)+'|'+datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S%f'),
					user=user,
					tab_id=tab_id,
					url = str(req.url),
					method=req.method,
					user_agent=get_user_agent(req),
					user_ip=get_user_ip(req),
					is_nginx=get_is_nginx(req)
				)
				user_request=db.merge(user_request)
				db.flush()		
				db.commit()

	# Si hi ha un error redireccionem altra vegada al login
	if len(my_error.strip())>0:		
		a=fh.RedirectResponse(redirection, status_code=303)
		print('enviem a  /softprop/login perque: ' + my_error)	
		return a, my_error  

	return '',''








Publicado por en
Etiquetas: , , , , , , , , , , , , , , , , , , ,

No hay comentarios :

Publicar un comentario

Suscribirse a: Enviar comentarios ( Atom )