Installation
$ source venv/bin/activate # On Windows use `venv\Scripts\activate`
$ pip install lila-framework
$ lila-init
$ python app.py #Or python3 app.py
# Import necessary modules and routes
from core.app import App
from routes.routes import routes
from routes.api import routes as api_routes
# Import environment variables for host and port
from core.env import PORT, HOST
import itertools
import uvicorn
import asyncio
# Optionally, uncomment the following imports for database migrations and connections.
# You can enable database migration or connection setup if necessary.
# from database.migrations import migrate
# from database.connections import connection
# Combine application and API routes into a single list.
# This combines both regular app routes and API routes for easy use.
all_routes = list(itertools.chain(routes, api_routes))
# Initialize the app with debugging enabled and combined routes.
# The app is initialized with debug mode on and the routes to handle the requests.
app = App(debug=True, routes=all_routes)
# Uncomment the following lines for CORS configuration if needed:
# CORS (Cross-Origin Resource Sharing) configuration is optional.
# cors={
# "origin": ["*"],
# "allow_credentials": True,
# "allow_methods":["*"],
# "allow_headers": ["*"]
# }
# app = App(debug=True, routes=all_routes, cors=cors)
# Main asynchronous function to run the application.
# This function starts the application server asynchronously.
async def main():
# Uncomment the following line to execute database migrations.
# This ensures that the database schema is up-to-date before starting the app.
# migrations = await migrate(connection)
# Start the Uvicorn server with the app instance.
# The app is served by Uvicorn, which is an ASGI server.
uvicorn.run("app:app.start", host=HOST, port=PORT, reload=True)
# Entry point of the script to run the main async function.
# This is where the execution of the app begins.
if __name__ == "__main__":
asyncio.run(main())
Routes are the access points to the application. In Lila, routes are defined in the
routes
directory (by default, but you can place them wherever you want) and
imported into
app.py
for use. Routes can be configured to handle HTTP requests, API
methods, and
more.
In addition to JSONResponse you can use HTMLResponse, RedirectResponse and PlainTextResponse, or StreamingResponse , to transmit data in real-time (useful for streaming video/audio or large responses).
Below is an example of how routes are defined in Lila:
#Import from core JSONResponse .
from core.responses import JSONResponse
# Manages routes for API endpoints.
from core.routing import Router
# Initializes the router instance to handle API routes.
router = Router()
# Defines a simple API route that supports the GET method.
@router.route(path='/api', methods=['GET'])
async def api(request: Request):
"""Api function"""
# English: Returns a simple JSON response for API verification.
return JSONResponse({'api': True})
In this function, we receive a parameter via
the URL using {param}
.
If the parameter is not provided,
it defaults to 'default'
.
The response is a JSON containing the received value.
@router.route(path='/url_with_param/{param}', methods=['GET'])
async def param(request: Request):
param = request.path_params.get('param', 'default')
return JSONResponse({"received_param": param})
Or you canl also do it like this
@router.route(path='/url_with_param/?query_param', methods=['GET','POST'])
async def query_param(request: Request):
query_param= request.query_params.get('query_param', 'query_param')
return JSONResponse({"received_param": query_params})
from core.responses import JSONResponse # Simplifies sending JSON responses.
from core.routing import Router # Manages API routes.
from core.request import Request # Handles HTTP requests in the application.
from pydantic import EmailStr, BaseModel # Validates and parses data models for input validation.
from core.helpers import get_user_by_id_and_token
from middlewares.middlewares import validate_token
router = Router()
Middlewares allow intercepting requests before they reach the main logic of the API.
In this example, we use @validate_token
to validate a JWT token in the
request
header.
@router.route(path='/api/token', methods=['GET','POST'])
@validate_token # Middleware to validate JWT token.
async def api_token(request: Request):
return JSONResponse({'api': True})
Pydantic allows defining data models that automatically validate user input.
Additionally, specifying a model in the route generates automatic documentation in
/docs
.
from pydantic import EmailStr, BaseModel
class ExampleModel(BaseModel):
email: EmailStr # Ensures a valid email.
password: str # String for password.
@router.route(path='/api/example', methods=['POST'], model=ExampleModel)
async def login(request: Request):
body = await request.json()
try:
input = ExampleModel(**body) # Automatic validation with Pydantic.
except Exception as e:
return JSONResponse({"success": False, "msg": f"Invalid JSON Body: {e}"}, status_code=400)
return JSONResponse({"email": input.email, "password": input.password})
Thanks to the integration with Pydantic, the API documentation is generated automatically
and is
accessible from
/docs
. You can also generate an OpenAPI JSON file for external tools.
router.swagger_ui() # Enables Swagger UI for API documentation.
router.openapi_json() # Generates OpenAPI JSON for external tools.
To use the routes defined in the router, you need to obtain them with
router.get_routes()
and import them into app.py
.
routes = router.get_routes() # Retrieves all defined routes.
To load static files(js,css,etc) you can use the mount()
method.
What will be received as parameters that trade by default
path: str = '/public', directory: str = 'static', name: str = 'static'
from core.routing import Router
# Creating an instance of the Router to define the routes, if it was not previously created in the file
router = Router()
# Mounting the static files in the 'static' folder, url ='/public' by default
router.mount()
In Lila, you can use Jinja2 to render HTML with server-side data or Lila JS to create reactive client-side components, or even combine both approaches according to your needs.
Lila offers three approaches to build your application:
Full server-side rendering using Jinja2 templates with context data.
<p>{{ translate['Welcome'] }} {{ user.name }}</p>
Advantages: Perfect SEO, fast initial load, simple.
Fully reactive single-page application on the client side.
<span data-bind="message"></span>
<input data-model="username">
Advantages: Fluid user experience, no page reloads.
Combine Jinja2 for base structure and Lila JS for interactive parts.
<div data-component="InteractivePart"></div>
<p>{{ static_content }}</p>
Advantages: The best of both worlds.
In Lila, Jinja2 is used by default to render HTML templates and send them to the
client.
With context
, you can pass information such as translations, data,
values, lists, dictionaries or whatever you need.
The parameters that the render
function from core.templates
can receive are:
request: Request,
template: str,
context: dict = {},
theme_: bool = True,
translate: bool = True,
files_translate: list = []
Lila JS is a minimalist library to create reactive interfaces with vanilla JavaScript.
State is a special object that automatically updates the UI when it changes.
state: () => ({
email: '',
password: ''
})
Bidirectional binding between inputs and state using data-model
.
<input data-model="email">
<span data-bind="email"></span>
Reusable units with their own state, template and actions.
App.createComponent('Login', {
template: 'login-template',
state: () => ({...}),
actions: {...}
});
Client-side navigation between components.
App.addRoute('/login', 'Login');
handleRouting();
This example shows how to combine Jinja2 for the base structure and Lila JS for interactivity.
<!DOCTYPE html>
<html lang="{{lang}}">
<head>
<meta charset="UTF-8">
<title>{{title}}</title>
<link rel="stylesheet" href="/public/css/styles.css" />
<script src="/public/js/lila.js"></script>
</head>
<body>
<main id="app-lila" class="container"></main>
{% include 'auth/login.html' with context %}
{% include 'auth/register.html' with context %}
{% include 'auth/footer.html' with context %}
<script>
handleRouting();
</script>
</body>
</html>
Explanation: This file serves as the base layout using Jinja2. It includes:
{{lang}}
and {{title}}
app-lila
where components will be mountedhandleRouting()
<template data-template="login-template">
<div class="flex center">
<article class="shadow rounded mx-m">
<h1 class="flex center">{{translate['login']}}</h1>
<form data-action="loginFetch">
<div class="input-icon">
<i class="icon-email"></i>
<input type="email" data-model="email" placeholder="Email" required />
</div>
<div class="input-icon">
<i class="icon-lock"></i>
<input type="password" data-model="password"
placeholder="{{translate['password']}}" required />
</div>
<button type="submit">{{translate['login']}}</button>
</form>
<a href="/register" data-link>{{translate['register']}}</a>
</article>
</div>
<div data-component="Footer"></div>
</template>
<script>
App.createComponent('Login', {
template: 'login-template',
state: () => ({
email: '',
password: ''
}),
actions: {
loginFetch: async ({ state, event }) => {
event.preventDefault();
const data = { email: state.email, password: state.password };
alert('Login: ' + JSON.stringify(data));
}
}
});
App.addRoute('/login', 'Login');
App.addRoute('*', 'Login'); // Default route
</script>
Key Features:
data-model
automatically sync with state.data-action
binds the form to the
loginFetch
method.
data-component="Footer"
.
data-link
navigate without
page reload.{{translate}}
for
internationalized texts.
<template data-template="register-template">
<div class="flex center container">
<article class="shadow rounded mx-m">
<h1>{{translate['register']}}</h1>
<form data-action="registerFetch">
<input type="text" data-model="name" placeholder="{{translate['name']}}">
<input type="email" data-model="email" placeholder="Email">
<input type="password" data-model="password" placeholder="{{translate['password']}}">
<input type="password" data-model="password_2" placeholder="{{translate['confirm_password']}}">
<button type="submit">{{translate['register']}}</button>
</form>
<div>
<p>Current values:</p>
<p>Name: <span data-bind="name"></span></p>
<p>Email: <span data-bind="email"></span></p>
</div>
<a href="/login" data-link>{{translate['login']}}</a>
</article>
</div>
</template>
<script>
App.createComponent('Register', {
template: 'register-template',
state: () => ({
name: '',
email: '',
password: '',
password_2: ''
}),
actions: {
registerFetch: async ({ state, event }) => {
event.preventDefault();
if (state.password !== state.password_2) {
alert("{{translate['passwords_dont_match']}}");
return;
}
const data = {
name: state.name,
email: state.email,
password: state.password
};
// Send data to server...
}
}
});
App.addRoute('/register', 'Register');
</script>
Key Points:
data-bind
<template data-template="footer-template">
<footer class="mt-4 container">
<div class="flex between">
<a href="/set-language/es">Español</a>
<a href="/set-language/en">English</a>
</div>
</footer>
</template>
<script>
App.createComponent('Footer', {
template: 'footer-template'
});
</script>
Notes:
Combines Jinja2's initial speed with SPA fluidity.
Critical content is server-rendered for search engine bots.
Start with Jinja2 and add interactivity where needed.
No complex configuration or heavy dependencies.
Attribute/Function | Description | Example |
---|---|---|
data-component |
Mounts a component inside another | <div data-component="Footer"></div> |
data-bind |
Displays a state property value | <span data-bind="email"></span> |
data-model |
Two-way binding for inputs | <input data-model="username"> |
data-action |
Binds an action to an event | <button data-action="submit"> |
data-link |
Navigation between routes without reload | <a href="/about" data-link> |
App.createComponent |
Creates a new component | App.createComponent('Login', {...}); |
App.addRoute |
Defines a route for a component | App.addRoute('/login', 'Login'); |
handleRouting() |
Initializes the routing system | <script>handleRouting();</script> |
The upload
helper provides a complete solution for handling file
uploads in your application,
including validation, security checks, and translation support.
This helper automatically handles:
The upload
function accepts these parameters:
request: Request # Starlette request object
name_file: str | list # Field name for file upload (default: 'file')
UPLOAD_DIR: str # Directory to save files (default: 'uploads')
ALLOWED_EXTENSIONS: set # Allowed file extensions (default: {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'})
MAX_FILE_SIZE: int # Maximum file size in bytes (default: 10MB)
from core.helpers import upload
@router.route("/upload", methods=["POST"])
async def uploadFile(request: Request):
response = await upload(request=request, name_file="file")
return response
<form onsubmit="upload(event);">
<fieldset>
<input type="file" name="file" required>
</fieldset>
<button type="submit">
<i class="icon-check-circle"></i>
Upload Files
</button>
</form>
<script>
async function upload(event) {
event.preventDefault();
const formElement = event.target;
const formData = new FormData(formElement);
try {
const response = await fetch('/upload', {
method: 'POST',
body: formData,
});
const result = await response.json();
if (!response.ok) throw new Error(result.message);
alert('File uploaded successfully!');
} catch (error) {
alert(error.message);
}
}
</script>
The helper returns a JSONResponse with this structure:
{
"file": "/uploads/filename.ext",
"success": true,
"message": "File uploaded successfully"
}
{
"error": true,
"success": false,
"message": "Error message in user's language"
}
To render markdown
files, the
renderMarkdown
function is used, which receives the following parameters:
request,file : str , base_path:str ='templates/markdown/',css_files : list = [],js_files:list=[],picocss : bool =False
file
is the name of the file, starting from the base_path
of
the
directory
located in templates/markdown/
, for example: index.md
.
It will look for an index.md
file in the markdown directory inside the
templates
folder
(templates/markdown/index.md).
css_files
and js_files
are lists of CSS and JS files that will
be
loaded in the
generated HTML file.
picocss
is a boolean indicating whether the PicoCSS CSS file should be
loaded.
Below is an example of how markdown
files are rendered in
Lilac:
from core.templates import renderMarkdown
@router.route(path='/markdown', methods=['GET'])
async def home(request: Request):
#Define a list of CSS files to include in the response
css = ["/public/css/styles.css"]
#Renders a markdown file with PicoCSS styling
response = renderMarkdown(request=request, file='example', css_files=css, picocss=True)
return response
Translations are used to internationalize an application and display content in different
languages. In Lila, translations are stored in the locales
directory and
can be dynamically loaded into the application.
The render
method accepts a lang_default
parameter to force a
specific language for the rendered template:
# Force Spanish language
@router.route(path="/es", methods=["GET"])
async def home(request: Request):
response = render(
request=request,
template="index",
lang_default="es" # Forces Spanish translations
)
return response
# Force English language
@router.route(path="/en", methods=["GET"])
async def home(request: Request):
response = render(
request=request,
template="index",
lang_default="en" # Forces English translations
)
return response
To load a locale file, use the translate
function from
core.helpers
. You can access translations using:
translate
- returns a dictionary with all translationstranslate_
- returns a specific translation (returns original text if
not found)Example translation file (locales/translations.json
):
{
"Send": {
"es": "Enviar",
"en": "Send"
},
"Cancel": {
"es": "Cancelar",
"en": "Cancel"
},
"Accept": {
"es": "Aceptar",
"en": "Accept"
},
"Email": {
"es": "Email",
"en": "Email"
},
"Name": {
"es": "Nombre",
"en": "Name"
},
"Back": {
"es": "Volver",
"en": "Back"
},
"Hi": {
"es": "Hola",
"en": "Hi"
}
}
Get a specific translation:
from core.helpers import translate_
msg_error_login = translate_(
key="Incorrect email or password",
request=request,
file_name="guest"
)
Get all translations from a file:
from core.helpers import translate
all_translations = translate(
request=request,
file_name="guest"
)
SQLAlchemy as the default ORM for database management. SQLAlchemy allows creating database models, executing queries, and handling migrations efficiently.
The `Base` class, imported from `core.database`, serves as the foundation for all models . Models inherit from `Base` to define database tables with SQLAlchemy.
from sqlalchemy import Table,Column,Integer,String,TIMESTAMP
from core.database import Base
from database.connections import connection
class User(Base):
__tablename__='users'
id = Column(Integer, primary_key=True,autoincrement=True)
name=Column( String(length=50), nullable=False)
email=Column( String(length=50), unique=True)
password=Column(String(length=150), nullable=False)
token=Column(String(length=150), nullable=False)
active=Column( Integer, nullable=False,default=1)
created_at=Column( TIMESTAMP)
#Example of how to use SQLAlchemy to make queries to the database
def get_all(select: str = "id,email,name", limit: int = 1000) -> list:
query = f"SELECT {select} FROM users WHERE active =1 LIMIT {limit}"
result = connection.query(query=query,return_rows=True)#Return rows
return result
#Example of how to use SQLAlchemy to make queries to the database
def get_by_id(id: int, select="id,email,name") -> dict:
query = f"SELECT {select} FROM users WHERE id = :id AND active = 1 LIMIT 1"
params = {"id": id}
row = connection.query(query=query, params=params,return_row=True)#Retrurn row
return row
#Example using ORM abstraction in SQLAlchemy
@classmethod
def get_all_orm(cls, db: Session, limit: int = 1000):
result = db.query(cls).filter(cls.active == 1).limit(limit).all()
return result
#Example of how to use the class to make queries to the database
users = User.get_all()
user = User.get_by_id(1)
For details on SQLAlchemy, visit the official documentation: SQLAlchemy Documentation.
middlewares
directory (can be
modified to any file and/or directory).
Middlewares can be used for tasks such as authentication, logging, and error handling.
By default, Lila comes with 3 middlewares to start any application. Middlewares can be
used with decorators @my_middleware
.
login_required
, to validate that you have a signed session, for the 'auth'
key that is passed as a parameter to be able to modify it as you wish.
If this session is not found, it redirects to the URL that is passed as a parameter, by
default, it is "/login"
.
Otherwise, it will continue its course executing the route or function.
Then we have session_active
, which is used to verify if you have an active
session.
It will redirect to the URL that is received as a parameter, by default, it is
"/dashboard"
.
The third is validate_token
, which is used to validate a JWT token thanks
to the get_token
helpers imported in
from core.helpers import get_token
.
from core.session import Session
from core.responses import RedirectResponse,JSONResponse
from core.request import Request
from functools import wraps
from core.helpers import get_token
def login_required(func,key:str='auth',url_return='/login'):
@wraps(func)
async def wrapper(request, *args, **kwargs):
session_data= Session.unsign(key=key,request=request)
if not session_data:
return RedirectResponse(url=url_return)
return await func(request,*args,**kwargs)
return wrapper
def session_active(func,key:str='auth',url_return:str ='/dashboard'):
@wraps(func)
async def wrapper(request,*args,**kwargs):
session_data= Session.unsign(key=key,request=request)
if session_data:
return RedirectResponse(url=url_return)
return await func(request,*args,**kwargs)
return wrapper
def validate_token(func):
@wraps(func)
async def wrapper(request: Request, *args, **kwargs):
await check_token(request=request)
return await func(request, *args, **kwargs)
return wrapper
async def check_token(request:Request):
token = request.headers.get('Authorization')
if not token:
return JSONResponse({'session':False,'message': 'Invalid token'},status_code=401)
token = get_token(token=token)
if isinstance(token,JSONResponse):
return token
#Middleware to validate the JWT Token.
@router.route(path='/api/token', methods=['GET','POST'])
@validate_token #Middleware
async def api_token(request: Request):
"""Api Token function"""
print(get_user_by_id_and_token(request=request))
return JSONResponse({'api': True})
#Middleware to validate session active
@router.route(path='/dashboard', methods=['GET'])
@login_required #Middleware
async def dashboard(request: Request):
response = render(request=request, template='dashboard', files_translate=['authenticated'])
return response
#Middleware validate if user get session active (if user get session redirect '/dashboard')
@router.route(path='/login', methods=['GET'])
@session_active #Middleware
async def login(request: Request):
response = render(request=request, template='auth/login', files_translate=['guest'])
return response
Lila Framework includes a built-in ErrorHandlerMiddleware
that not only
handles unhandled exceptions but also provides robust security checks to protect your
application from malicious requests. This middleware is designed to block suspicious
IPs, URLs, and sensitive paths, ensuring that your application remains secure.
/etc/passwd
,
.env
, and others.
.php
,
.asp
, .jsp
, and .aspx
.
"http"
in query parameters or body content.
The middleware uses three JSON files located in the security
directory:
blocked_ips.json
: Stores blocked IPs with their expiration time.blocked_urls.json
: Stores blocked URLs with their expiration time.sensitive_paths.json
: Stores a list of sensitive paths to block.If these files do not exist, they are automatically created and initialized with default values:
[
"/etc/passwd",
".env",
"wp-content",
"docker/.env",
"owa/auth/logon.aspx",
"containers/json",
"models",
"autodiscover/autodiscover.json",
"heapdump",
"actuator/heapdump",
"cgi-bin/vitogate.cgi",
"CFIDE/wizards/common/utils.cfc",
"var/www/html/.env",
"home/user/.muttrc",
"usr/local/spool/mail/root",
"etc/postfix/master.cf"
]
The ErrorHandlerMiddleware
is automatically applied to all requests. You
can customize its behavior by modifying the JSON files in the security
directory.
from starlette.middleware.base import BaseHTTPMiddleware
from core.responses import JSONResponse, HTMLResponse
from core.request import Request
from core.logger import Logger
from datetime import datetime, timedelta
import json
import os
def load_blocked_data(file_path, default_value):
try:
if not os.path.exists(file_path):
with open(file_path, "w") as file:
json.dump(default_value, file, indent=4)
return default_value
with open(file_path, "r") as file:
content = file.read().strip()
if not content:
with open(file_path, "w") as file:
json.dump(default_value, file, indent=4)
return default_value
try:
return json.loads(content)
except json.JSONDecodeError:
with open(file_path, "w") as file:
json.dump(default_value, file, indent=4)
return default_value
except Exception as e:
Logger.error(f"Error loading {file_path}: {str(e)}")
return default_value
def save_blocked_data(file_path, data):
try:
with open(file_path, "w") as file:
json.dump(data, file, indent=4)
except Exception as e:
Logger.error(f"Error saving {file_path}: {str(e)}")
async def is_blocked(blocked_data, key, request: Request):
if key in blocked_data:
expiration_time = datetime.fromisoformat(blocked_data[key]["expiration_time"])
if datetime.now() < expiration_time:
req = await Logger.request(request=request)
Logger.warning(f"Blocked: {key} \n {req}")
return True
return False
class ErrorHandlerMiddleware(BaseHTTPMiddleware):
def __init__(
self,
app,
blocked_ips_file="security/blocked_ips.json",
blocked_urls_file="security/blocked_urls.json",
sensitive_paths_file="security/sensitive_paths.json",
):
super().__init__(app)
self.blocked_ips_file = blocked_ips_file
self.blocked_urls_file = blocked_urls_file
self.sensitive_paths_file = sensitive_paths_file
self.blocked_ips = load_blocked_data(blocked_ips_file, default_value={})
self.blocked_urls = load_blocked_data(blocked_urls_file, default_value={})
self.sensitive_paths = load_blocked_data(sensitive_paths_file, default_value=[])
async def dispatch(self, request, call_next):
try:
client_ip = request.client.host
url_path = request.url.path
query_params = str(request.query_params)
body = await request.body()
if await is_blocked(self.blocked_ips, client_ip, request=request):
return HTMLResponse(
content="Access Denied
Your IP has been temporarily blocked.
",
status_code=403,
)
if await is_blocked(self.blocked_urls, url_path, request=request):
return HTMLResponse(
content="Access Denied
This URL has been temporarily blocked.
",
status_code=403,
)
malicious_extensions = [".php", ".asp", ".jsp", ".aspx"]
if any(ext in url_path for ext in malicious_extensions):
self.blocked_ips[client_ip] = {
"expiration_time": (datetime.now() + timedelta(hours=6)).isoformat()
}
save_blocked_data(self.blocked_ips_file, self.blocked_ips)
return HTMLResponse(
content="Access Denied
Malicious URL detected.
",
status_code=403,
)
if "http" in query_params or "http" in str(body):
self.blocked_ips[client_ip] = {
"expiration_time": (datetime.now() + timedelta(hours=6)).isoformat()
}
save_blocked_data(self.blocked_ips_file, self.blocked_ips)
return HTMLResponse(
content="Access Denied
Malicious query parameters detected.
",
status_code=403,
)
if any(path in url_path or path in str(body) for path in self.sensitive_paths):
self.blocked_ips[client_ip] = {
"expiration_time": (datetime.now() + timedelta(hours=6)).isoformat()
}
save_blocked_data(self.blocked_ips_file, self.blocked_ips)
return HTMLResponse(
content="Access Denied
Sensitive path detected.
",
status_code=403,
)
Logger.info(await Logger.request(request=request))
response = await call_next(request)
return response
except Exception as e:
Logger.error(f"Unhandled Error: {str(e)}")
return JSONResponse(
{"error": "Internal Server Error", "success": False}, status_code=500
)
To use connections, you need to import the Database
class from
core.database. With
that you can connect to your database, which can be SQLite, MySLQ, PostgreSQL or
whatever you
want to configure.
Below we leave you the example of how to connect. The connection will close automatically
after
being used, so you can use it as in this example in the variable connection
from core.database import Database
#SQLite
#Example connection to a sqlite database
config = {"type":"sqlite","database":"test"} #test.db
connection = Database(config=config)
connection.connect()
#MySql
#Example connection to a mysql database
config = {"type":"mysql","host":"127.0.0.1","user":"root","password":"password","database":"db_test","auto_commit":True}
connection = Database(config=config)
connection.connect()
mysql_connection = connection
In Lila Framework, database migrations can be managed using SQLAlchemy and Lila's configuration to make migrations as easy as possible. The framework now supports command-line migrations through Typer, providing a more intuitive and flexible way to manage your database schema.
There are two main ways to define database tables:
This method manually defines the table structure using SQLAlchemy's Table object.
from sqlalchemy import Table, Column, Integer, String, TIMESTAMP
from database.connections import connection
import typer
import asyncio
app = typer.Typer()
# Example of creating migrations for 'users'
table_users = Table(
'users', connection.metadata,
Column('id', Integer, primary_key=True, autoincrement=True),
Column('name', String(length=50), nullable=False),
Column('email', String(length=50), unique=True),
Column('password', String(length=150), nullable=False),
Column('token', String(length=150), nullable=False),
Column('active', Integer, default=1, nullable=False),
Column('created_at', TIMESTAMP),
)
async def migrate_async(connection, refresh: bool = False) -> bool:
try:
if refresh:
connection.metadata.drop_all(connection.engine)
connection.prepare_migrate([table_users]) # for tables
connection.migrate()
print("Migrations completed")
return True
except RuntimeError as e:
print(e)
return False
@app.command()
def migrate(refresh: bool = False):
"""Run database migrations"""
success = asyncio.run(migrate_async(connection, refresh))
if not success:
raise typer.Exit(code=1)
if __name__ == "__main__":
app()
This approach defines database tables as Python classes that inherit from "Base". This is the recommended method as it provides more structure and ORM capabilities.
from core.database import Base
from sqlalchemy import Column, Integer, String, TIMESTAMP
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(length=50), nullable=False)
email = Column(String(length=50), unique=True)
password = Column(String(length=150), nullable=False)
token = Column(String(length=150), nullable=False)
active = Column(Integer, nullable=False, default=1)
created_at = Column(TIMESTAMP)
from database.connections import connection
from core.database import Base # Import Base for migrations in models
from models.user import User # Import models for migrations
import typer
import asyncio
app = typer.Typer()
async def migrate_async(connection, refresh: bool = False) -> bool:
try:
if refresh:
connection.metadata.drop_all(connection.engine)
connection.migrate(use_base=True) # for models
print("Migrations completed")
return True
except RuntimeError as e:
print(e)
return False
@app.command()
def migrate(refresh: bool = False):
"""Run database migrations"""
success = asyncio.run(migrate_async(connection, refresh))
if not success:
raise typer.Exit(code=1)
if __name__ == "__main__":
app()
To execute migrations, use the following command in your terminal:
# Basic migration
python -m database.migrations
# Refresh all tables (drop and recreate)
python -m database.migrations --refresh
migrate
: Run the database migrations--refresh
: Optional flag to drop and recreate all tablesNote: When using Models, make sure to import all your model classes in the migrations file so SQLAlchemy can detect them for migrations.
Thanks to the combination of SQLAlchemy and Pydantic models, it is possible to perform data validations and execute structured queries for API generation.
Additionally, you can integrate custom middlewares to validate tokens, manage sessions, or process requests. With just a few lines of code, you can generate a fully documented REST API CRUD.
If you haven't already, enable migrations when you start the server
.
By default it uses SQLite, it will create a database file lila.sqlite
in
the
project root.
from database.migrations import migrate
from database.connections import connection
async def main():
# execute migrations ,for app
migrations = await migrate(connection)
uvicorn.run("app:app.start", host=HOST, port=PORT, reload=True)
if __name__ == "__main__":
asyncio.run(main())
from core.request import Request
from core.responses import JSONResponse
from core.routing import Router
from pydantic import EmailStr, BaseModel
from middlewares.middlewares import validate_token, check_token, check_session
from database.connections import connection # Database connection with SQLAlchemy
from models.user import User # SQLAlchemy 'User' model
router = Router()# Initialize the router instance for managing API routes.
# Pydantic model for validations when creating or modifying a user.
class UserModel(BaseModel):
email: EmailStr
name: str
token: str
password: str
# Middleware definitions for CRUD operations
middlewares_user = {
"get": [],
"post": [],
"get_id": [],
"put": [],
"delete": [check_session, check_token],#Example middlewares for web session 'check_session' and jwt 'check_token'
}
# Automatically generate CRUD with validations and configurations
router.rest_crud_generate(
connection=connection, # Database connection
model_sql=User, # SQLAlchemy model
model_pydantic=UserModel, # Pydantic model
select=["name", "email", "id", "created_at", "active"], # Fields to select in queries
delete_logic=True, # Enables logical deletion (updates 'active = 0' instead of deleting records)
active=True, # Automatically filters active records ('active = 1')
middlewares=middlewares_user, # Custom middlewares for each CRUD action
)
You can create your own middlewares and pass them as a list to customize
security and validations for each rest_crud_generate
operation.
To generate the documentation, always remember to
run it after the routes router.swagger_ui()
and
router.openapi_json()
rest_crud_generate
Below are the parameters that this function accepts to generate CRUD automatically:
def rest_crud_generate(
self,
connection,
model_sql,
model_pydantic: Type[BaseModel],
select: Optional[List[str]] = None,
columns: Optional[List[str]] = None,
active: bool = False,
delete_logic: bool = False,
middlewares: dict = None,
jsonresponse_prefix:str='',#Return with prefix list o dict 'data' first key
user_id_session:bool| str=False #Example 'user id' to validate in query with where 'user_id'= id session_user
) :
Below is an example of the generated documentation for the
rest_crud_generate
function:
Go to http://127.0.0.1:8001/docs
, or as configured in your .env file (by
HOST and
PORT).
In this example, we did it with 'users', but you can apply it however you want according
to your
logic ,'products','stores',etc. Even modifying the core in core/routing.py
.
For both the 'POST' or 'PUT' method function, If the framework detects that you pass data in the request body such as: 'password' , it will automatically encode it with argon2 to make it secure. Body example :
{
"email":"example@example.com",
"name":"name",
"password":"my_password_secret"
}
Then, if you pass 'token' or 'hash', with the helper
function
generate_token_value
, it automatically generates a token, which will be saved in the database as the 'token'
column
with the value generated by the function
.
Body example :
{
"email":"example@example.com",
"name":"name",
"password":"my_password_secret",
"token":""
}
With 'created_at' or 'created_date' , it will save the date and time of the moment, as long as that field exists in the database table. Body example :
{
"email":"example@example.com",
"name":"name",
"password":"my_password_secret",
"token":"",
"created_at":""
}
For the 'PUT', 'GET' (get_id) or 'DELETE' methods
It is optional according to the logic of each REST API, you can pass it as
query string
, user_id
or id_user
, an example would be GET, PUT or DELETE as a method
to the url http://127.0.0.1:8001/api/products/1?user_id=20
Where it validates that the product ID '1' exists but also that it belongs to the user ID '20' .
The Admin
module allows you to manage an admin panel for your application.
It includes authentication, model management, system metrics, and more. This panel is
highly customizable and integrates easily with your application.
The admin panel is now more modular and flexible. All admin-related components
(templates, routes, and configuration)
are located in the admin
folder, making it easier to customize and extend.
Logs
.
To use the admin panel, you need to import the Admin
class from
admin.routes
and pass it
an optional list of models you want to manage. Each model must implement a
get_all
method to be displayed in the admin panel.
# English: Here we activate the admin panel with default settings.
# Español: Aquí activamos el panel de administrador con configuraciones predeterminadas.
from admin.routes import Admin
from models.user import User
admin_routes=Admin(models=[User])
all_routes = list(itertools.chain(routes, api_routes,admin_routes))
Before using the admin panel, you need to run the migrations:
python -m database.migrations
Admin users can be created via command line with customizable parameters:
# Default usage (random password generated)
python -m admin.create_admin
# With custom username and password
python -m admin.create_admin --user myadmin --password mysecurepassword
The admin panel automatically generates the following routes:
/admin/login
(GET/POST)/admin/logout
(GET)/admin
(GET)/admin/{model_plural}
(GET)
To protect routes and ensure only authenticated administrators can access them, use the
@admin_required
decorator from core/admin.py
.
from core.admin import admin_required
@router.route(path="/admin", methods=["GET"])
@admin_required
async def admin_route(request: Request):
menu_html = menu(models=models)
return await admin_dashboard(request=request, menu=menu_html)
The admin panel displays real-time metrics, including:
These metrics are updated every 10 seconds.
In Lila, we use a middleware that you can enable or disable if you want to use
Logs
for info, warnings, or errors in your application.
The middleware is located in core/middleware.py
and is added to the
application with:
app.add_middleware(ErrorHandlerMiddleware)
.
This helps generate logs that you can view in the admin panel, organized by date (in folders) and type.