| Name | FastSQLA JSON |
| Version |
0.2.4
JSON |
| download |
| home_page | None |
| Summary | SQLAlchemy extension for FastAPI that supports asynchronous sessions and includes built-in pagination. |
| upload_time | 2025-01-27 19:29:00 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.12 |
| license | MIT License |
| keywords |
fastapi
sqlalchemy
asyncio
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# 🚀 FastSQLA
[](https://pypi.org/project/FastSQLA/)
[](https://conventionalcommits.org)
[](https://codecov.io/gh/hadrien/fastsqla)
`FastSQLA` is an [`SQLAlchemy`] extension for [`FastAPI`].
It supports asynchronous `SQLAlchemy` sessions and includes built-in custimizable
pagination.
## Features
<details>
<summary>Automatic SQLAlchemy configuration at app startup.</summary>
Using [`FastAPI` Lifespan](https://fastapi.tiangolo.com/advanced/events/#lifespan):
```python
from fastapi import FastAPI
from fastsqla import lifespan
app = FastAPI(lifespan=lifespan)
```
</details>
<details>
<summary>Async SQLAlchemy session as a FastAPI dependency.</summary>
```python
...
from fastsqla import Session
from sqlalchemy import select
...
@app.get("/heros")
async def get_heros(session:Session):
stmt = select(...)
result = await session.execute(stmt)
...
```
</details>
<details>
<summary>Built-in pagination.</summary>
```python
...
from fastsqla import Page, Paginate
from sqlalchemy import select
...
@app.get("/heros", response_model=Page[HeroModel])
async def get_heros(paginate:Paginate):
return paginate(select(Hero))
```
</details>
<details>
<summary>Allows pagination customization.</summary>
```python
...
from fastapi import new_pagination
...
Paginate = new_pagination(min_page_size=5, max_page_size=500)
@app.get("/heros", response_model=Page[HeroModel])
async def get_heros(paginate:Paginate):
return paginate(select(Hero))
```
</details>
And more ...
<!-- <details><summary></summary></details> -->
## Installing
Using [uv](https://docs.astral.sh/uv/):
```bash
uv add fastsqla
```
Using [pip](https://pip.pypa.io/):
```
pip install fastsqla
```
## Quick Example
```python
# example.py
from http import HTTPStatus
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Mapped, mapped_column
from fastsqla import Base, Item, Page, Paginate, Session, lifespan
app = FastAPI(lifespan=lifespan)
class Hero(Base):
__tablename__ = "hero"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(unique=True)
secret_identity: Mapped[str]
class HeroBase(BaseModel):
name: str
secret_identity: str
class HeroModel(HeroBase):
model_config = ConfigDict(from_attributes=True)
id: int
@app.get("/heros", response_model=Page[HeroModel])
async def list_users(paginate: Paginate):
return await paginate(select(Hero))
@app.get("/heros/{hero_id}", response_model=Item[HeroModel])
async def get_user(hero_id: int, session: Session):
hero = await session.get(Hero, hero_id)
if hero is None:
raise HTTPException(HTTPStatus.NOT_FOUND, "Hero not found")
return {"data": hero}
@app.post("/heros", response_model=Item[HeroModel])
async def create_user(new_hero: HeroBase, session: Session):
hero = Hero(**new_hero.model_dump())
session.add(hero)
try:
await session.flush()
except IntegrityError:
raise HTTPException(HTTPStatus.CONFLICT, "Duplicate hero name")
return {"data": hero}
```
> [!NOTE]
> Sqlite is used for the sake of the example.
> FastSQLA is compatible with all async db drivers that SQLAlchemy is compatible with.
<details>
<summary>Create an <code>sqlite3</code> db:</summary>
```bash
sqlite3 db.sqlite <<EOF
CREATE TABLE hero (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL UNIQUE, -- Hero name (e.g., Superman)
secret_identity TEXT NOT NULL -- Secret identity (e.g., Clark Kent)
);
-- Insert heroes with hero name and secret identity
INSERT INTO hero (name, secret_identity) VALUES ('Superman', 'Clark Kent');
INSERT INTO hero (name, secret_identity) VALUES ('Batman', 'Bruce Wayne');
INSERT INTO hero (name, secret_identity) VALUES ('Wonder Woman', 'Diana Prince');
INSERT INTO hero (name, secret_identity) VALUES ('Iron Man', 'Tony Stark');
INSERT INTO hero (name, secret_identity) VALUES ('Spider-Man', 'Peter Parker');
INSERT INTO hero (name, secret_identity) VALUES ('Captain America', 'Steve Rogers');
INSERT INTO hero (name, secret_identity) VALUES ('Black Widow', 'Natasha Romanoff');
INSERT INTO hero (name, secret_identity) VALUES ('Thor', 'Thor Odinson');
INSERT INTO hero (name, secret_identity) VALUES ('Scarlet Witch', 'Wanda Maximoff');
INSERT INTO hero (name, secret_identity) VALUES ('Doctor Strange', 'Stephen Strange');
INSERT INTO hero (name, secret_identity) VALUES ('The Flash', 'Barry Allen');
INSERT INTO hero (name, secret_identity) VALUES ('Green Lantern', 'Hal Jordan');
EOF
```
</details>
<details>
<summary>Install dependencies & run the app</summary>
```bash
pip install uvicorn aiosqlite fastsqla
sqlalchemy_url=sqlite+aiosqlite:///db.sqlite?check_same_thread=false uvicorn example:app
```
</details>
Execute `GET /heros?offset=10`:
```bash
curl -X 'GET' \
'http://127.0.0.1:8000/heros?offset=10&limit=10' \
-H 'accept: application/json'
```
Returns:
```json
{
"data": [
{
"name": "The Flash",
"secret_identity": "Barry Allen",
"id": 11
},
{
"name": "Green Lantern",
"secret_identity": "Hal Jordan",
"id": 12
}
],
"meta": {
"offset": 10,
"total_items": 12,
"total_pages": 2,
"page_number": 2
}
}
```
[`FastAPI`]: https://fastapi.tiangolo.com/
[`SQLAlchemy`]: http://sqlalchemy.org/
Raw data
{
"_id": null,
"home_page": null,
"name": "FastSQLA",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "FastAPI, SQLAlchemy, AsyncIO",
"author": null,
"author_email": "Hadrien David <bonjour@hadriendavid.com>",
"download_url": "https://files.pythonhosted.org/packages/5f/91/ad1b298d7be88c30964c45490160de1f2bc1505ccbd21b144e1b3b73d20e/fastsqla-0.2.4.tar.gz",
"platform": null,
"description": "# \ud83d\ude80 FastSQLA\n\n[](https://pypi.org/project/FastSQLA/)\n[](https://conventionalcommits.org)\n[](https://codecov.io/gh/hadrien/fastsqla)\n\n`FastSQLA` is an [`SQLAlchemy`] extension for [`FastAPI`].\nIt supports asynchronous `SQLAlchemy` sessions and includes built-in custimizable\npagination.\n\n## Features\n\n<details>\n <summary>Automatic SQLAlchemy configuration at app startup.</summary>\n\n Using [`FastAPI` Lifespan](https://fastapi.tiangolo.com/advanced/events/#lifespan):\n```python\nfrom fastapi import FastAPI\nfrom fastsqla import lifespan\n\napp = FastAPI(lifespan=lifespan)\n```\n</details>\n<details>\n <summary>Async SQLAlchemy session as a FastAPI dependency.</summary>\n\n```python\n...\nfrom fastsqla import Session\nfrom sqlalchemy import select\n...\n\n@app.get(\"/heros\")\nasync def get_heros(session:Session):\n stmt = select(...)\n result = await session.execute(stmt)\n ...\n```\n</details>\n<details>\n <summary>Built-in pagination.</summary>\n\n```python\n...\nfrom fastsqla import Page, Paginate\nfrom sqlalchemy import select\n...\n\n@app.get(\"/heros\", response_model=Page[HeroModel])\nasync def get_heros(paginate:Paginate):\n return paginate(select(Hero))\n```\n</details>\n<details>\n <summary>Allows pagination customization.</summary>\n\n```python\n...\nfrom fastapi import new_pagination\n...\n\nPaginate = new_pagination(min_page_size=5, max_page_size=500)\n\n@app.get(\"/heros\", response_model=Page[HeroModel])\nasync def get_heros(paginate:Paginate):\n return paginate(select(Hero))\n```\n</details>\n\nAnd more ...\n<!-- <details><summary></summary></details> -->\n\n## Installing\n\nUsing [uv](https://docs.astral.sh/uv/):\n```bash\nuv add fastsqla\n```\n\nUsing [pip](https://pip.pypa.io/):\n```\npip install fastsqla\n```\n\n## Quick Example\n\n```python\n# example.py\nfrom http import HTTPStatus\n\nfrom fastapi import FastAPI, HTTPException\nfrom pydantic import BaseModel, ConfigDict\nfrom sqlalchemy import select\nfrom sqlalchemy.exc import IntegrityError\nfrom sqlalchemy.orm import Mapped, mapped_column\n\nfrom fastsqla import Base, Item, Page, Paginate, Session, lifespan\n\napp = FastAPI(lifespan=lifespan)\n\n\nclass Hero(Base):\n __tablename__ = \"hero\"\n id: Mapped[int] = mapped_column(primary_key=True)\n name: Mapped[str] = mapped_column(unique=True)\n secret_identity: Mapped[str]\n\n\nclass HeroBase(BaseModel):\n name: str\n secret_identity: str\n\n\nclass HeroModel(HeroBase):\n model_config = ConfigDict(from_attributes=True)\n id: int\n\n\n@app.get(\"/heros\", response_model=Page[HeroModel])\nasync def list_users(paginate: Paginate):\n return await paginate(select(Hero))\n\n\n@app.get(\"/heros/{hero_id}\", response_model=Item[HeroModel])\nasync def get_user(hero_id: int, session: Session):\n hero = await session.get(Hero, hero_id)\n if hero is None:\n raise HTTPException(HTTPStatus.NOT_FOUND, \"Hero not found\")\n return {\"data\": hero}\n\n\n@app.post(\"/heros\", response_model=Item[HeroModel])\nasync def create_user(new_hero: HeroBase, session: Session):\n hero = Hero(**new_hero.model_dump())\n session.add(hero)\n try:\n await session.flush()\n except IntegrityError:\n raise HTTPException(HTTPStatus.CONFLICT, \"Duplicate hero name\")\n return {\"data\": hero}\n```\n\n> [!NOTE]\n> Sqlite is used for the sake of the example.\n> FastSQLA is compatible with all async db drivers that SQLAlchemy is compatible with.\n\n<details>\n <summary>Create an <code>sqlite3</code> db:</summary>\n\n```bash\nsqlite3 db.sqlite <<EOF\nCREATE TABLE hero (\n id INTEGER PRIMARY KEY AUTOINCREMENT,\n name TEXT NOT NULL UNIQUE, -- Hero name (e.g., Superman)\n secret_identity TEXT NOT NULL -- Secret identity (e.g., Clark Kent)\n);\n\n-- Insert heroes with hero name and secret identity\nINSERT INTO hero (name, secret_identity) VALUES ('Superman', 'Clark Kent');\nINSERT INTO hero (name, secret_identity) VALUES ('Batman', 'Bruce Wayne');\nINSERT INTO hero (name, secret_identity) VALUES ('Wonder Woman', 'Diana Prince');\nINSERT INTO hero (name, secret_identity) VALUES ('Iron Man', 'Tony Stark');\nINSERT INTO hero (name, secret_identity) VALUES ('Spider-Man', 'Peter Parker');\nINSERT INTO hero (name, secret_identity) VALUES ('Captain America', 'Steve Rogers');\nINSERT INTO hero (name, secret_identity) VALUES ('Black Widow', 'Natasha Romanoff');\nINSERT INTO hero (name, secret_identity) VALUES ('Thor', 'Thor Odinson');\nINSERT INTO hero (name, secret_identity) VALUES ('Scarlet Witch', 'Wanda Maximoff');\nINSERT INTO hero (name, secret_identity) VALUES ('Doctor Strange', 'Stephen Strange');\nINSERT INTO hero (name, secret_identity) VALUES ('The Flash', 'Barry Allen');\nINSERT INTO hero (name, secret_identity) VALUES ('Green Lantern', 'Hal Jordan');\nEOF\n```\n\n</details>\n\n<details>\n <summary>Install dependencies & run the app</summary>\n\n```bash\npip install uvicorn aiosqlite fastsqla\nsqlalchemy_url=sqlite+aiosqlite:///db.sqlite?check_same_thread=false uvicorn example:app\n```\n\n</details>\n\nExecute `GET /heros?offset=10`:\n\n```bash\ncurl -X 'GET' \\\n'http://127.0.0.1:8000/heros?offset=10&limit=10' \\\n-H 'accept: application/json'\n```\nReturns:\n```json\n{\n \"data\": [\n {\n \"name\": \"The Flash\",\n \"secret_identity\": \"Barry Allen\",\n \"id\": 11\n },\n {\n \"name\": \"Green Lantern\",\n \"secret_identity\": \"Hal Jordan\",\n \"id\": 12\n }\n ],\n \"meta\": {\n \"offset\": 10,\n \"total_items\": 12,\n \"total_pages\": 2,\n \"page_number\": 2\n }\n}\n```\n\n[`FastAPI`]: https://fastapi.tiangolo.com/\n[`SQLAlchemy`]: http://sqlalchemy.org/\n",
"bugtrack_url": null,
"license": "MIT License",
"summary": "SQLAlchemy extension for FastAPI that supports asynchronous sessions and includes built-in pagination.",
"version": "0.2.4",
"project_urls": {
"Changelog": "https://github.com/hadrien/fastsqla/releases",
"Documentation": "https://github.com/hadrien/fastsqla",
"Homepage": "https://github.com/hadrien/fastsqla",
"Issues": "https://github.com/hadrien/fastsqla/issues",
"Repository": "https://github.com/hadrien/fastsqla"
},
"split_keywords": [
"fastapi",
" sqlalchemy",
" asyncio"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "f713039bde32ef2c49e122d3e8a114e304f4a46f5a6de340d74550965b0edddc",
"md5": "d4bb6e27a61e893a7348a3738deddd17",
"sha256": "c474edd62e20bb247d295edb10db7e0429c4fbe8c45ae935a1bea2c239517c2a"
},
"downloads": -1,
"filename": "FastSQLA-0.2.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d4bb6e27a61e893a7348a3738deddd17",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 6103,
"upload_time": "2025-01-27T19:28:58",
"upload_time_iso_8601": "2025-01-27T19:28:58.880767Z",
"url": "https://files.pythonhosted.org/packages/f7/13/039bde32ef2c49e122d3e8a114e304f4a46f5a6de340d74550965b0edddc/FastSQLA-0.2.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "5f91ad1b298d7be88c30964c45490160de1f2bc1505ccbd21b144e1b3b73d20e",
"md5": "126a9677e4b72c47db1e93d96d0ff4f8",
"sha256": "7d1f93b0b9721eb70e04d2f5e7b696e5fe61ea45ee149828d2c06af7b5551b8a"
},
"downloads": -1,
"filename": "fastsqla-0.2.4.tar.gz",
"has_sig": false,
"md5_digest": "126a9677e4b72c47db1e93d96d0ff4f8",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 6374,
"upload_time": "2025-01-27T19:29:00",
"upload_time_iso_8601": "2025-01-27T19:29:00.751072Z",
"url": "https://files.pythonhosted.org/packages/5f/91/ad1b298d7be88c30964c45490160de1f2bc1505ccbd21b144e1b3b73d20e/fastsqla-0.2.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-01-27 19:29:00",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "hadrien",
"github_project": "fastsqla",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "fastsqla"
}
