I Created SQLModel-Crud-Utilities So CRUD Operations Can Be Easier
A set of CRUD (Create, Read, Update, Delete) utilities designed to streamline and expedite common database operations when using SQLModel
In the ever-evolving world of web development, efficiency and simplicity stand as paramount pillars for developers. Python, with its robust libraries and frameworks such as SQLAlchemy and FastAPI, has become a favorite for creating powerful web applications. However, while SQLAlchemy provides an amazing ORM for database interactions, managing CRUD (Create, Read, Update, Delete) operations can still become cumbersome as applications scale. Enter sqlmodel-crud-utilities, a game-changer for developers looking to streamline these operations.
Introducing SQLModel-CRUD-Utilities
Available for download on PyPI, this library is designed to simplify CRUD operations when combined with SQLModel and FastAPI. SQLModel is a library that offers an elegant way to work with SQL databases through Python objects, while FastAPI is a modern web framework for building APIs with Python based on standard RESTful principles.
The sqlmodel-crud-utilities package provides a set of utilities to make the development of RESTful APIs both cleaner and less repetitive. Using these utilities, developers can reduce boilerplate code associated with CRUD operations, allowing them to focus more on the core logic of their applications.
Benefits and Features
This package stands out because of its comprehensive features that address real-world database operation needs:
- Sync & Async Support: Offers parallel abilities to manage operations both synchronously (sqlmodel_crud_utils.sync) and asynchronously (sqlmodel_crud_utils.a_sync).
- Comprehensive CRUD Functions: Implements high-level operations like
get_one_or_create,get_row, andwrite_row,which cater to frequent data handling use-cases with elegance. - Relationship Loading: Facilitates eager loading and lazy loading relationships easily within your queries.
With built-in pagination, error-handling with logging via loguru, and support for dialect-specific upsert operations, this utility shines in both functionality and reliability.
Ease of Implementation
By providing a structured and intuitive interface for CRUD operations, developers can quickly integrate database functionalities without delving deep into the complexities of SQLModel or SQLAlchemy.
Consistency Across Projects
Development teams can maintain consistent database interaction patterns across different projects, fostering uniform code practices and facilitating easier onboarding of new developers.
Performance Optimizations
The library has been built keeping in mind the need for optimization, hence it leverages some of FastAPI’s performance efficiencies inherently.
Enhanced Code Maintainability
With blowflake templates for CRUD operations readily available, maintaining and upgrading code becomes less of a hassle.
Customizable and Flexible
Despite its utility for common use cases, it provides room for customization to accommodate unique business logic.
Getting Started
To start using sqlmodel-crud-utilities, you'll need a basic setup with FastAPI and SQLModel. Here's a quick guide on how you can integrate the library into your project:
Installation:
Simply install the package via pip or uv:
pip install sqlmodel-crud-utilities
# or
uv pip install sql-model-crud-utilities
Then, ensure that you declare the SQL_DIALECT environmental variable, so that the dependency can make the necessary imports for upsert commands. Alternatively, you can declare this in an .env file and load the value using a dependency like python-dotenv.
export SQL_DIALECT=postgresql # or sqlite, mysql etc
Demonstration with Real-World Application
To demonstrate an actual use case for the module, the following example will be provided. Applications such as e-commerce platforms, content management systems, or any scenario where database interactions are inevitable, can benefit enormously from using this package. Developers can achieve faster implementation times and reduce error potential in CRUD operations.
Here is a simple FastAPI application to manage products for an e-commerce service. This is a very naive bootstrapped version, intended to demonstrate the built-in requirements for a CRUD API application:
# app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
# Define the FastAPI app
app = FastAPI()
# Define the product model
class Product(BaseModel):
id: int
name: str
description: str
price: float
quantity: int
# Simulate a database with an in-memory list
products_db = []
# Create a product
@app.post("/products/", response_model=Product)
def create_product(product: Product):
products_db.append(product)
return product
# Read all products
@app.get("/products/", response_model=List[Product])
def list_products():
return products_db
# Read a single product by ID
@app.get("/products/{product_id}", response_model=Product)
def read_product(product_id: int):
for product in products_db:
if product.id == product_id:
return product
raise HTTPException(status_code=404, detail="Product not found")
# Update a product
@app.put("/products/{product_id}", response_model=Product)
def update_product(product_id: int, product: Product):
for idx, existing_product in enumerate(products_db):
if existing_product.id == product_id:
products_db[idx] = product
return product
raise HTTPException(status_code=404, detail="Product not found")
# Delete a product
@app.delete("/products/{product_id}", response_model=Product)
def delete_product(product_id: int):
for idx, product in enumerate(products_db):
if product.id == product_id:
del products_db[idx]
return product
raise HTTPException(status_code=404, detail="Product not found")
This example illustrates the need to write boilerplate code that either pairs with an ORM or ODM to collect the necessary results via search parameters or key ID lookup. The example uses a list object in place of an actual database, but it serves the intended demonstration purpose.
We would ordinarily write query commands to process searches with a database session, such as with SQLAlchemy. Doing this across multiple applications can require significant boilerplate code, and projects can increasingly get out of date as application portfolios scale. Instead, with sqlmodel-crud-utilities, we do this:
# app.py
from fastapi import FastAPI, HTTPException
from typing import List
from sqlmodel import Session, SQLModel, create_engine, Field
from sqlmodel_crud_utils.sync import get_one_or_create, get_row, get_rows, update_row, delete_row
# Database URL setup
DATABASE_URL = "sqlite:///./mydatabase.db"
engine = create_engine(DATABASE_URL)
SQLModel.metadata.create_all(engine)
# Define the FastAPI app
app = FastAPI()
# Define the product model
class Product(SQLModel, table=True):
id: int | None = Field(default=None, primary_key=True)
name: str
description: str
price: float
quantity: int
# Create a product
@app.post("/products/", response_model=Product)
def create_product(product: Product):
with Session(engine) as session:
instance, created = get_one_or_create(session_inst=session, model=Product,
id=product.id,
create_method_kwargs=product.dict())
if not created:
raise HTTPException(status_code=400, detail="Product already exists")
return instance
# Read all products
@app.get("/products/", response_model=List[Product])
def list_products():
with Session(engine) as session:
success, products = get_rows(session_inst=session, model=Product)
if not success:
raise HTTPException(status_code=404, detail="No products found")
return products
# Read a single product by ID
@app.get("/products/{product_id}", response_model=Product)
def read_product(product_id: int):
with Session(engine) as session:
success, product = get_row(session_inst=session, model=Product, id=product_id)
if not success:
raise HTTPException(status_code=404, detail="Product not found")
return product
# Update a product
@app.put("/products/{product_id}", response_model=Product)
def update_product(product_id: int, product: Product):
with Session(engine) as session:
existing_product = read_product(product_id)
if not existing_product:
raise HTTPException(status_code=404, detail="Product not found")
update_success, updated_product = update_row(session_inst=session, model=Product,
id=product_id, update_dict=product.dict())
if not update_success:
raise HTTPException(status_code=400, detail="Failed to update product")
return updated_product
# Delete a product
@app.delete("/products/{product_id}", response_model=Product)
def delete_product(product_id: int):
with Session(engine) as session:
success, product = get_row(session_inst=session, model=Product, id=product_id)
if not success:
raise HTTPException(status_code=404, detail="Product not found")
delete_success = delete_row(session_inst=session, model=Product, id=product_id)
if not delete_success:
raise HTTPException(status_code=400, detail="Failed to delete product")
return product
Key Changes:
- Created a database engine and setup the Product model with SQLModel-based ORM support.
- Implemented sqlmodel-crud-utilities functions (
get_one_or_create,get_row,get_rows,update_row,delete_row) for CRUD operations, thereby eliminating the manual handling of product records. - Used session.commit() where necessary to ensure changes are saved to the database.
Also note that we do not need to declare session.commit() or otherwise manage session, as that is taken care of for us by the functions.
This setup integrates SQLModel with the CRUD utility library for efficient database operations in your FastAPI application.
The Upsert Command
sqlmodel-crud-utilities includes a bulk_upsert_mappings command to process multiple data rows elegantly. Most simply, the generated SQL command will attempt to create new rows for an associated datatable with a fall-back to update existing rows based on declared unique field values. The use-case can include ETL workflows where data from various repositories pushes new data values to the backend, necessitating a need to elegantly handle conflicts.
# adding to app.py
# Create or update a product (upsert operation)
@app.post("/upsert-product/", response_model=Product)
def upsert_products(products: list[Product]):
with Session(engine) as session:
success, upserted_product = bulk_upsert_mappings(
payload=products,
session_inst=session,
model=Product,
# We'll not declare the `pk_fields` argument, as we just need to check against the `id` field.
)
if not success:
raise HTTPException(status_code=400, detail="Failed to upsert product")
return upserted_product
Testing Data
To test the upsert endpoint, you would create an array of JSON data that adheres to the Product schema and await the 201 response. You can do this either within the /docs API testing endpoint or via your preferred HTTP client, such as CURL.
An example would be this:
[
{
"id": 1,
"name": "Test Product",
"description": "A sample product for testing upsert functionality",
"price": 19.99,
"quantity": 100
}
]
Running the Application
To run the FastAPI application, use Uvicorn:
uvicorn main:app --reload
Contributing and Looking Ahead
There's a tremendous opportunity to contribute and refine sqlmodel-crud-utilities as it progresses through its beta phase into a 1.0 release. Community feedback is invaluable, and contributors can help shape a robust ecosystem around this utility.
Looking ahead, the roadmap includes enhancing continuous integration via GitHub Actions, achieving 100% test coverage with Pytest, and solidifying a stable release.
Conclusion
As we usher into an era where frameworks and libraries define the landscape of software development, sqlmodel-crud-utilities stands out as a crucial tool that simplifies complex operations, enhances productivity, and improves code quality. For developers invested in Python's modern web frameworks, utilizing this library could be a significant step toward building scalable and maintainable APIs.
With this powerful utility in your toolkit, create with clarity and efficiency, channeling your efforts to innovate and enhance the features of your application rather than getting bogged down by mundane database interactions.