# Fluent Alchemy
輕鬆流暢地使用 SQLAclhemy
- [Installation](#installation)
- [Quick start](#quick-start)
- [Features](#features)
- [Mixins](#mixins)
- [Examples](#examples)
## Installation
```shell
pip insall fluent-alchemy
```
## Quick start
1. 宣告 Models 並繼承 `ActiveRecord`
```python
# models.py
from sqlalchemy.orm import DeclarativeBase
from fluent_alchemy import ActiveRecord
class Base(DeclarativeBase, ActiveRecord):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(BigInteger(), primary_key=True)
email: Mapped[str] = mapped_column(String(), unique=True)
password: Mapped[str] = mapped_column(String())
name: Mapped[str] = mapped_column(String(50))
state: Mapped[bool] = mapped_column(Boolean())
```
2. 建立 SQLAclhemy Engine, 並指派給 `ActiveRecord` 內的 session handler
```python
from sqlalchemy import create_engine
from models import Base
engine = create_engine("sqlite://:memory:", echo=True)
Base.set_engine(engine)
```
3. 開始使用 model 操作 database !
```python
from models import User
users = User.all()
```
4. 使用完畢後，釋放 Session 資源
```python
from models import Base
Base.remove_scoped_session()
```
## Features
### Active Record
利用 `QueryBuilder` 來處理 SQLAlchemy 的 `select()` query statement
- Create
```python
from models import User
user = User.create(
name="Justin",
email="corey97@example.org",
password="NioWe9Wn#+"
)
# or
user = User(
name="Justin",
email="corey97@example.org",
password="NioWe9Wn#+"
)
user.save()
```
- Read
1. Find by id
```python
user = User.find(1)
```
2. 只回傳特定欄位
```python
user = User.select(User.id, User.name, User.email).first()
```
3. 透過 `where` 增加查詢條件
```python
user = User.where(User.email == "corey97@example.org").first()
```
```python
users = User.where(User.state.is_(True)).get()
```
- Update
```python
user = User.find(1)
user.passwod = "6xjVAY$p&D"
user.save()
```
- Delete
```python
user = User.find(1)
user.delete()
```
- Pagenate
```python
# setting page number and rows count per page
pagination = User.paginate(page=1, per_page=15)
"""
{
"total": 100,
"per_page": 15,
"current_page": 1,
"last_page": 7,
"data": [ ... ], # ussers
}
"""
```
## Mixins
### [`TimestampMixin`](./fluent_alchemy/mixins/timestamp.py)
讓指定的 Model class 繼承 `TimestampMixin`，讓該 Model 補上 `created_at`, `updated_at` 欄位。
```python
from fluent_alchemy import ActiveRecord, TimestampMixin
class Base(DeclarativeBase, ActiveRecord):
pass
class User(Base, TimestampMixin):
__tablename__ = "users"
...
###
user = User.find(1)
print(user.created_at)
print(user.updated_at)
```
### [`SoftDeleteMixin`](./fluent_alchemy/mixins/softdelete.py)
讓指定的 Model class 繼承 `SoftDeleteMixin`，就可以讓該 Model 擁有 Soft delete 的能力。
```python
from sqlalchemy.orm import DeclarativeBase
from fluent_alchemy import ActiveRecord, SoftDeleteMixin
class Base(DeclarativeBase, ActiveRecord):
pass
class User(Base, SoftDeleteMixin):
__tablename__ = "users"
id: Mapped[int] = mapped_column(BigInteger(), primary_key=True)
email: Mapped[str] = mapped_column(String(), unique=True)
password: Mapped[str] = mapped_column(String())
name: Mapped[str] = mapped_column(String(50))
state: Mapped[bool] = mapped_column(Boolean())
```
`SoftDeleteMixin` 會自動補上 `deleted_at` 欄位，依此欄位來處理 soft delete 的資料。
```python
deleted_at: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP(), nullable=True)
```
設定完成後，之後對此 Model 進行 Query 時，會在 statement 內的 WHERE 條件自動加上 `deleted_at IS NULL`。
#### 查詢已被標記刪除的資料
```python
users_deleted = User.where(...).get(with_trashed=True)
```
#### 強制刪除 (Force delete)
```python
user = User.find(1)
user.delete(force=True)
```
## Examples
### 在 FastAPI 內使用
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from app.models import BaseModel, User
def close_scoped_session():
"""
Remove the scoped session at the enf of every request.
"""
yield
BaseModel.remove_scoped_session()
@asynccontextmanager
async def lifespan(app: FastAPI):
"""
Set engine to the ScopedSessionHandler when FastAPI app started.
"""
BaseModel.set_engine(engine)
yield
app = FastAPI(
title="MyApp",
dependencies=[Depends(close_scoped_session)],
lifespan=lifespan
)
@app.get("/users")
def index():
return User.all()
```
Raw data
{
"_id": null,
"home_page": "https://github.com/turisesonia/fluent-alchemy",
"name": "fluent-alchemy",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": "sqlalchemy, SQLAlchemy, activerecord, Active Record",
"author": "Sam Yao",
"author_email": "turisesonia@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/e7/27/4bca5f51e6ccadbf686f02ce1814b52d2458548f366934c4ab61a755dd0e/fluent_alchemy-0.0.2.tar.gz",
"platform": null,
"description": "# Fluent Alchemy\n\n\u8f15\u9b06\u6d41\u66a2\u5730\u4f7f\u7528 SQLAclhemy\n\n- [Installation](#installation)\n- [Quick start](#quick-start)\n- [Features](#features)\n- [Mixins](#mixins)\n- [Examples](#examples)\n\n## Installation\n\n```shell\npip insall fluent-alchemy\n```\n\n## Quick start\n\n1. \u5ba3\u544a Models \u4e26\u7e7c\u627f `ActiveRecord`\n\n```python\n# models.py\nfrom sqlalchemy.orm import DeclarativeBase\nfrom fluent_alchemy import ActiveRecord\n\nclass Base(DeclarativeBase, ActiveRecord):\n pass\n\nclass User(Base):\n __tablename__ = \"users\"\n\n id: Mapped[int] = mapped_column(BigInteger(), primary_key=True)\n email: Mapped[str] = mapped_column(String(), unique=True)\n password: Mapped[str] = mapped_column(String())\n name: Mapped[str] = mapped_column(String(50))\n state: Mapped[bool] = mapped_column(Boolean())\n\n```\n\n2. \u5efa\u7acb SQLAclhemy Engine, \u4e26\u6307\u6d3e\u7d66 `ActiveRecord` \u5167\u7684 session handler\n\n```python\nfrom sqlalchemy import create_engine\nfrom models import Base\n\nengine = create_engine(\"sqlite://:memory:\", echo=True)\n\nBase.set_engine(engine)\n```\n\n3. \u958b\u59cb\u4f7f\u7528 model \u64cd\u4f5c database !\n\n```python\nfrom models import User\n\nusers = User.all()\n```\n\n4. \u4f7f\u7528\u5b8c\u7562\u5f8c\uff0c\u91cb\u653e Session \u8cc7\u6e90\n\n```python\nfrom models import Base\n\nBase.remove_scoped_session()\n```\n\n## Features\n\n### Active Record\n\u5229\u7528 `QueryBuilder` \u4f86\u8655\u7406 SQLAlchemy \u7684 `select()` query statement\n\n- Create\n\n ```python\n from models import User\n\n user = User.create(\n name=\"Justin\",\n email=\"corey97@example.org\",\n password=\"NioWe9Wn#+\"\n )\n\n # or\n\n user = User(\n name=\"Justin\",\n email=\"corey97@example.org\",\n password=\"NioWe9Wn#+\"\n )\n user.save()\n ```\n\n- Read\n 1. Find by id\n\n ```python\n user = User.find(1)\n ```\n\n 2. \u53ea\u56de\u50b3\u7279\u5b9a\u6b04\u4f4d\n\n ```python\n user = User.select(User.id, User.name, User.email).first()\n ```\n\n 3. \u900f\u904e `where` \u589e\u52a0\u67e5\u8a62\u689d\u4ef6\n\n ```python\n user = User.where(User.email == \"corey97@example.org\").first()\n ```\n\n ```python\n users = User.where(User.state.is_(True)).get()\n ```\n\n- Update\n\n ```python\n user = User.find(1)\n user.passwod = \"6xjVAY$p&D\"\n\n user.save()\n ```\n\n- Delete\n\n ```python\n user = User.find(1)\n user.delete()\n ```\n\n- Pagenate\n\n ```python\n # setting page number and rows count per page\n pagination = User.paginate(page=1, per_page=15)\n\n \"\"\"\n {\n \"total\": 100,\n \"per_page\": 15,\n \"current_page\": 1,\n \"last_page\": 7,\n \"data\": [ ... ], # ussers\n }\n \"\"\"\n ```\n\n\n## Mixins\n\n### [`TimestampMixin`](./fluent_alchemy/mixins/timestamp.py)\n\n\u8b93\u6307\u5b9a\u7684 Model class \u7e7c\u627f `TimestampMixin`\uff0c\u8b93\u8a72 Model \u88dc\u4e0a `created_at`, `updated_at` \u6b04\u4f4d\u3002\n\n```python\nfrom fluent_alchemy import ActiveRecord, TimestampMixin\n\nclass Base(DeclarativeBase, ActiveRecord):\n pass\n\nclass User(Base, TimestampMixin):\n __tablename__ = \"users\"\n ...\n\n###\nuser = User.find(1)\n\nprint(user.created_at)\nprint(user.updated_at)\n```\n\n\n### [`SoftDeleteMixin`](./fluent_alchemy/mixins/softdelete.py)\n\n\u8b93\u6307\u5b9a\u7684 Model class \u7e7c\u627f `SoftDeleteMixin`\uff0c\u5c31\u53ef\u4ee5\u8b93\u8a72 Model \u64c1\u6709 Soft delete \u7684\u80fd\u529b\u3002\n\n```python\nfrom sqlalchemy.orm import DeclarativeBase\nfrom fluent_alchemy import ActiveRecord, SoftDeleteMixin\n\nclass Base(DeclarativeBase, ActiveRecord):\n pass\n\nclass User(Base, SoftDeleteMixin):\n __tablename__ = \"users\"\n\n id: Mapped[int] = mapped_column(BigInteger(), primary_key=True)\n email: Mapped[str] = mapped_column(String(), unique=True)\n password: Mapped[str] = mapped_column(String())\n name: Mapped[str] = mapped_column(String(50))\n state: Mapped[bool] = mapped_column(Boolean())\n```\n\n`SoftDeleteMixin` \u6703\u81ea\u52d5\u88dc\u4e0a `deleted_at` \u6b04\u4f4d\uff0c\u4f9d\u6b64\u6b04\u4f4d\u4f86\u8655\u7406 soft delete \u7684\u8cc7\u6599\u3002\n\n```python\ndeleted_at: Mapped[Optional[datetime]] = mapped_column(TIMESTAMP(), nullable=True)\n```\n\n\u8a2d\u5b9a\u5b8c\u6210\u5f8c\uff0c\u4e4b\u5f8c\u5c0d\u6b64 Model \u9032\u884c Query \u6642\uff0c\u6703\u5728 statement \u5167\u7684 WHERE \u689d\u4ef6\u81ea\u52d5\u52a0\u4e0a `deleted_at IS NULL`\u3002\n\n#### \u67e5\u8a62\u5df2\u88ab\u6a19\u8a18\u522a\u9664\u7684\u8cc7\u6599\n```python\nusers_deleted = User.where(...).get(with_trashed=True)\n```\n\n#### \u5f37\u5236\u522a\u9664 (Force delete)\n```python\nuser = User.find(1)\n\nuser.delete(force=True)\n```\n\n## Examples\n\n### \u5728 FastAPI \u5167\u4f7f\u7528\n\n```python\n\nfrom contextlib import asynccontextmanager\nfrom fastapi import FastAPI, Depends\nfrom app.models import BaseModel, User\n\ndef close_scoped_session():\n \"\"\"\n Remove the scoped session at the enf of every request.\n \"\"\"\n yield\n BaseModel.remove_scoped_session()\n\n@asynccontextmanager\nasync def lifespan(app: FastAPI):\n \"\"\"\n Set engine to the ScopedSessionHandler when FastAPI app started.\n \"\"\"\n BaseModel.set_engine(engine)\n yield\n\napp = FastAPI(\n title=\"MyApp\",\n dependencies=[Depends(close_scoped_session)],\n lifespan=lifespan\n)\n\n@app.get(\"/users\")\ndef index():\n return User.all()\n\n```",
"bugtrack_url": null,
"license": "MIT",
"summary": "Use SQLAlchemy fluently",
"version": "0.0.2",
"project_urls": {
"Documentation": "https://github.com/turisesonia/fluent-alchemy",
"Homepage": "https://github.com/turisesonia/fluent-alchemy",
"Repository": "https://github.com/turisesonia/fluent-alchemy"
},
"split_keywords": [
"sqlalchemy",
" sqlalchemy",
" activerecord",
" active record"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "0bb8dd775f5300af462bdf4b18a8c352e3e6e654ee2142d6308c163fd6183a35",
"md5": "13f0c909c7e0139ba1f602d59cafae4a",
"sha256": "a70d3e0ca14696a36a40b17c62a3e7d8157bceb273c71bae4a959833ad383ff0"
},
"downloads": -1,
"filename": "fluent_alchemy-0.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "13f0c909c7e0139ba1f602d59cafae4a",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 10372,
"upload_time": "2024-04-09T07:37:04",
"upload_time_iso_8601": "2024-04-09T07:37:04.115621Z",
"url": "https://files.pythonhosted.org/packages/0b/b8/dd775f5300af462bdf4b18a8c352e3e6e654ee2142d6308c163fd6183a35/fluent_alchemy-0.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "e7274bca5f51e6ccadbf686f02ce1814b52d2458548f366934c4ab61a755dd0e",
"md5": "9883a1d4372a45cea28e084d94e2304c",
"sha256": "9ffb786ef63d397c879675c3715a5c253a3ec549b14069585a7ca3fbfff029a9"
},
"downloads": -1,
"filename": "fluent_alchemy-0.0.2.tar.gz",
"has_sig": false,
"md5_digest": "9883a1d4372a45cea28e084d94e2304c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 8446,
"upload_time": "2024-04-09T07:37:06",
"upload_time_iso_8601": "2024-04-09T07:37:06.110729Z",
"url": "https://files.pythonhosted.org/packages/e7/27/4bca5f51e6ccadbf686f02ce1814b52d2458548f366934c4ab61a755dd0e/fluent_alchemy-0.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-04-09 07:37:06",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "turisesonia",
"github_project": "fluent-alchemy",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "fluent-alchemy"
}
JSON
