### Install AirSQLModel with PostgreSQL Support Source: https://github.com/pydanny/airsqlmodel/blob/main/README.md Installs the AirSQLModel library with PostgreSQL support using the 'uv' package installer. Ensure you have 'uv' installed and configured. ```bash uv add "airsqlmodel[postgresql]" ``` -------------------------------- ### Install AirSQLModel with SQLite Support Source: https://github.com/pydanny/airsqlmodel/blob/main/README.md Installs the AirSQLModel library with SQLite support using the 'uv' package installer. Ensure you have 'uv' installed and configured. ```bash uv add "airsqlmodel[sqlite]" ``` -------------------------------- ### Get Asynchronous Session for Background Tasks Source: https://context7.com/pydanny/airsqlmodel/llms.txt Provides an asynchronous session generator for use in background tasks or standalone asynchronous functions. It manages the session lifecycle within an async context manager. ```python import airsqlmodel as sql from sqlmodel import select, SQLModel, Field class Task(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) title: str completed: bool = False async def complete_all_tasks(): # Use async context manager for session async with sql.get_async_session() as session: # Query incomplete tasks statement = select(Task).where(Task.completed == False) result = await session.execute(statement) tasks = result.scalars().all() # Mark all as completed for task in tasks: task.completed = True session.add(task) await session.commit() return len(tasks) ``` -------------------------------- ### Async Session Dependency in Air Views Source: https://context7.com/pydanny/airsqlmodel/llms.txt Demonstrates how to use an asynchronous database session as a dependency in Air views for automatic session management. This allows for easy querying and manipulation of data within web request handlers. It requires importing necessary modules from 'air' and 'airsqlmodel', and defining SQLModel models. ```python import air import airsqlmodel as sql from sqlmodel import SQLModel, Field, select from air import Request app = air.Air(lifespan=sql.create_async_db_lifespan()) class Hero(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) name: str secret_identity: str age: int | None = None @app.page async def heroes_list( request: Request, session: sql.AsyncSession = sql.async_session_dependency ): # Query all heroes statement = select(Hero) result = await session.execute(statement) heroes = result.scalars().all() return air.Main( air.H1("Heroes"), air.Ul(*[ air.Li(f"{hero.name} ({hero.secret_identity})") for hero in heroes ]) ) @app.post("/heroes") async def create_hero( request: Request, session: sql.AsyncSession = sql.async_session_dependency ): data = await request.json() hero = Hero(**data) session.add(hero) await session.commit() await session.refresh(hero) return {"id": hero.id, "name": hero.name} ``` -------------------------------- ### Configure Air App with Database Connection Source: https://context7.com/pydanny/airsqlmodel/llms.txt Sets up an Air web application with database connection management using environment variables and lifespan management. It defines a SQLModel for the 'User' table. ```python import air import airsqlmodel as sql from sqlmodel import SQLModel, Field # Set DATABASE_URL environment variable # export DATABASE_URL="postgresql://user:password@localhost/dbname" # or for SQLite: export DATABASE_URL="sqlite:///database.db" # Create Air app with database lifespan management app = air.Air(lifespan=sql.create_async_db_lifespan()) # Define a SQLModel table class User(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) name: str email: str active: bool = True ``` -------------------------------- ### Perform SQL Queries within Air Views Source: https://github.com/pydanny/airsqlmodel/blob/main/README.md Demonstrates how to make SQLModel queries inside Air views using the `async_session_dependency`. This dependency requires the `DATABASE_URL` environment variable to be set and provides an asynchronous database session. ```python import air import airsqlmodel as sql from sqlmodel import select # Assuming User model is defined elsewhere and accessible # from your_models import User app = air.Air(lifespan=sql.async_db_lifespan) @app.page async def index(request: Request, session: sql.AsyncSession = air.Depends(sql.async_session_dependency)): # Use the session to interact with the database result = await session.execute(select(User).where(User.name == "John")) user = result.scalars().first() return air.Main( air.H1("User Info"), air.P(f"Name: {user.name}"), air.P(f"Email: {user.email}"), ) ``` -------------------------------- ### Create Asynchronous Session Factory Source: https://context7.com/pydanny/airsqlmodel/llms.txt Generates an asynchronous session factory for database operations, suitable for use outside of Air views. It supports custom connection URLs and configuration for logging. ```python import airsqlmodel as sql from sqlmodel import SQLModel, Field, select class Product(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) name: str price: float async def process_products(): # Create session factory session_factory = await sql.create_async_session( url="sqlite+aiosqlite:///:memory:", echo=sql._EchoEnum.FALSE ) # Use session for database operations async with session_factory() as session: # Insert new product product = Product(name="Widget", price=19.99) session.add(product) await session.commit() await session.refresh(product) # Query products result = await session.execute(select(Product)) products = result.scalars().all() return products ``` -------------------------------- ### Create Asynchronous Database Engine Source: https://context7.com/pydanny/airsqlmodel/llms.txt Creates an asynchronous SQLAlchemy engine optimized for modern web applications, featuring connection pooling, pre-ping validation, and configurable options for URL, logging, and future features. ```python import airsqlmodel as sql # Create async engine with default settings async_engine = sql.create_async_engine() # Create async engine with custom configuration async_engine = sql.create_async_engine( url="postgresql+asyncpg://user:pass@localhost/db", echo=sql._EchoEnum.TRUE, future=sql._FutureEnum.TRUE, pool_pre_ping=sql._PoolPrePingEnum.TRUE ) # Use in async context async with async_engine.begin() as conn: await conn.run_sync(SQLModel.metadata.create_all) await async_engine.dispose() ``` -------------------------------- ### Retrieve Object or Return 404 Source: https://context7.com/pydanny/airsqlmodel/llms.txt Provides a utility function to fetch a database record by specified criteria or automatically return a 404 Not Found response. This is particularly useful for detail views where a resource might not exist. It requires the session, the model class, and the filtering conditions as arguments. ```python import air import airsqlmodel as sql from sqlmodel import SQLModel, Field from air import Request from fastapi import Depends app = air.Air(lifespan=sql.create_async_db_lifespan()) class Article(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) slug: str = Field(unique=True, index=True) title: str content: str published: bool = False @app.get("/articles/{slug}") async def article_detail( slug: str, session: sql.AsyncSession = sql.async_session_dependency ): # Get article by slug or raise 404 article = await sql.get_object_or_404( session, Article, Article.slug == slug, Article.published == True ) return air.layouts.mvpcss( air.H1(article.title), air.Article(air.P(article.content)) ) @app.get("/users/{user_id}/profile") async def user_profile( user_id: int, session: sql.AsyncSession = sql.async_session_dependency ): try: user = await sql.get_object_or_404( session, User, User.id == user_id, User.active == True ) return {"id": user.id, "name": user.name, "email": user.email} except air.exceptions.ObjectDoesNotExist as e: return {"error": "User not found"}, 404 ``` -------------------------------- ### Create Synchronous Database Engine Source: https://context7.com/pydanny/airsqlmodel/llms.txt Generates a synchronous SQLAlchemy engine for database operations, supporting default or custom connection URLs and optional logging of SQL statements. ```python import airsqlmodel as sql # Create synchronous engine with default DATABASE_URL engine = sql.create_sync_engine() # Create engine with custom URL and echo enabled engine = sql.create_sync_engine( url="sqlite:///myapp.db", echo=sql._EchoEnum.TRUE ) # Use with SQLModel metadata operations from sqlmodel import SQLModel SQLModel.metadata.create_all(engine) ``` -------------------------------- ### Create Async Database Lifespan Context Source: https://context7.com/pydanny/airsqlmodel/llms.txt Establishes an asynchronous database connection lifespan context manager for an Air application. This ensures database connections are properly managed throughout the application's life, preventing common timeout issues. It allows for custom database URLs, ensuring connections are made to the correct database instance. ```python import air import airsqlmodel as sql from sqlmodel import SQLModel, Field # Define your models class Customer(SQLModel, table=True): id: int | None = Field(default=None, primary_key=True) name: str email: str # Create lifespan with custom database URL lifespan = sql.create_async_db_lifespan( url="postgresql://user:password@localhost/production_db" ) # Initialize Air app with lifespan app = air.Air(lifespan=lifespan) @app.page async def customers( request: air.Request, session: sql.AsyncSession = sql.async_session_dependency ): from sqlmodel import select statement = select(Customer) result = await session.execute(statement) customers = result.scalars().all() return air.Main( air.H1("Customers"), air.Table( air.Tr(air.Th("ID"), air.Th("Name"), air.Th("Email")), *[ air.Tr( air.Td(str(c.id)), air.Td(c.name), air.Td(c.email) ) for c in customers ] ) ) ``` -------------------------------- ### Make SQLModel Queries Outside Air Views Source: https://github.com/pydanny/airsqlmodel/blob/main/README.md Shows how to obtain an asynchronous database session outside of Air views, such as in background tasks, using `airsqlmodel.get_async_session`. This allows for database interactions in various parts of your application. ```python import airsqlmodel as sql from sqlmodel import select # Assuming User model is defined elsewhere and accessible # from your_models import User async def some_background_task(): async with sql.get_async_session() as session: result = await session.execute(select(User).where(User.active == True)) active_users = result.scalars().all() # Do something with active_users ``` -------------------------------- ### Configure Air App with Database Lifespan Source: https://github.com/pydanny/airsqlmodel/blob/main/README.md Configures the Air application instance to use an asynchronous database lifespan, ensuring the database connection remains active. This is crucial for preventing connection expiration errors. ```python import air import airsqlmodel as sql app = air.Air(lifespan=sql.async_db_lifespan) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.