[](https://pypi.python.org/pypi/true-noorm/)
[](https://pypi.python.org/pypi/true-noorm/)
[](https://github.com/amaslyaev/noorm/blob/master/LICENSE)
[](https://codecov.io/gh/amaslyaev/noorm)
[](https://github.com/psf/black/)
## noorm
NoORM (Not only ORM) - make your database operations convenient and natural
## Install
**noorm** requires Python 3.10 or newer. Install it from PyPI:
```shell
$ pip install true-noorm
```
Please note that the correct name is "**true**-noorm".
## NoORM principles
1. It is not a holy war against ORM but "in addition to".
2. It is not one more "finally perfect" ORM. It is not an ORM at all. **No persistent objects, no "ideal" entities.**
3. It should be good for for medium-sized and big project.
4. Focus on developer experience.
5. Not only a set of helpers to write less code, but first of all, an approach that guides to more understandable, performant, scalable, robust, and maintainable solutions.
## Usage
Impotring `noorm` depends on DB you use in your project. Available options:
- `import noorm.sqlite3 as nm` – for SQLite
- `import noorm.aiosqlite as nm` – for asynchronous SQLite via **aiosqlite**
- `import noorm.psycopg2 as nm` – for synchronous Postgres via **psycopg2**
- `import noorm.asyncpg as nm` – for asynchronous Postgres via **asyncpg**
- `import noorm.pymysql as nm` – for synchronous MySQL/MariaDB via **PyMySQL**
- `import noorm.aiomysql as nm` – for asynchronous MySQL/MariaDB via **aiomysql**
- `import noorm.sqlalchemy_sync as nm` – for synchronous **SqlAlchemy**
- `import noorm.sqlalchemy_async as nm` – for asynchronous **SqlAlchemy**
Yes, using the SqlAlchemy ORM through NoORM is a nice idea.
After importing "nm" you use `@nm.sql_...` decorators to create functions that perform database operations. All other your application code uses these functions as a so-called "DB API layer". The decorators are:
- **@nm.sql_fetch_all** – to make a query and produce a list of objects of specified type. The query is usually SELECT, but it is also useful with data manipulations RETURNING data.
- **@nm.sql_one_or_none** – to make a query and produce one object or None if nothing is found.
- **@nm.sql_scalar_or_none** – to get a scalar or None if nothing is found.
- **@nm.sql_fetch_scalars** – to get a list of scalars.
- **@nm.sql_execute** – to execute something, usually INSERT, UPDATE, or DELETE.
- **@nm.sql_iterate** and **@nm.sql_iterate_scalars** – to make a query and iterate through results – objects or scalars respectively. Be careful with this features and, if possible, use `sql_fetch_all` and `sql_fetch_scalars` instead, because they give you less possibilites to shoot your leg. These functions are not implemented for **asyncpg**.
Usage of these decorators in different submodules and underlying databases might have own peculiarities, so check docstring documentation of the chosen "nm".
#### Example for SQLite through the sqlite3 standard library
Assume we have a **users** table with fields:
- Integer `id` (`rowid` in SQLite)
- String `username`
- String `email`
And an **orders** table:
- Integer `id` (`rowid` in SQLite)
- Integer `user_id` references to user id
- Datetime `order_date` (TEXT in SQLite)
- Decimal `amount` (TEXT in SQLite)
```python
from dataclasses import dataclass
from decimal import Decimal
from datetime import datetime
import sqlite3
import noorm.sqlite3 as nm
@dataclass
class DbUser: # When we need only basic info. Not a "model"! Just a dataclass!
id: int
username: str
email: str
@nm.sql_fetch_all(DbUser, "SELECT rowid AS id, username, email FROM users")
def get_all_users():
pass # no parameters, so just "pass"
@nm.sql_one_or_none(
DbUser, "SELECT rowid AS id, username, email FROM users WHERE rowid = :id"
)
def get_user_by_id(id_: int):
return nm.params(id=id_)
@dataclass
class DbUserWithOrdersSummary: # With additional info from `orders` table
id: int
username: str
sum_orders: Decimal | None # SQLite noorm can make decimal out of TEXT
first_order: datetime | None # and datatime too.
last_order: datetime | None
@nm.sql_fetch_all(
DbUserWithOrdersSummary,
"""SELECT
u.rowid AS id, u.username,
SUM(o.amount) AS sum_orders,
MIN(o.order_date) AS first_order, MAX(o.order_date) AS last_order
FROM users u
LEFT OUTER JOIN orders o ON o.user_id = u.rowid
GROUP BY u.rowid, u.username ORDER BY u.rowid
""",
)
def get_users_with_order_summary():
pass
def main():
with sqlite3.connect("data.sqlite") as conn:
# will use our DB API functions
for usr in get_all_users(conn):
print(usr)
print(f"{get_user_by_id(conn, 1)=}")
for usr_summary in get_users_with_order_summary(conn):
print(usr_summary)
```
#### Example for SQLite through SqlAlchemy
Will use asynchronous version of SqlAlchemy "nm".
```python
import sqlalchemy as sa
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
import noorm.sqlalchemy_async as nm
...
@nm.sql_fetch_all(DbUser)
def get_all_users(): # Notice no "async"
return sa.select(User.id, User.username, User.email)
@nm.sql_one_or_none(DbUser)
def get_user_by_id(id_: int):
return sa.select(User.id, User.username, User.email).where(User.id == id_)
@nm.sql_fetch_all(DbUserWithOrdersSummary)
def get_users_with_order_summary():
return (
sa.select(
User.id,
User.username,
sa.func.sum(Order.amount).label("sum_orders"),
sa.func.min(Order.order_date).label("first_order"),
sa.func.max(Order.order_date).label("last_order"),
)
.select_from(User) # `User` and `Order` are ORM model classes
.outerjoin(Order, Order.user_id == User.id)
.group_by(User.id)
.order_by(User.id)
)
async def main():
engine = create_async_engine("sqlite+aiosqlite:///data.sqlite")
async with AsyncSession(engine) as session:
for usr in await get_all_users(session): # Notice "await"
print(usr)
print(f"user_1={await get_user_by_id(session, 1)}")
for usr_summary in await get_users_with_order_summary(session):
print(usr_summary)
```
## Suggestions
1. Code structure:
- Avoid scattering DB API functions throughout the codebase. It's preferable to consolidate them in dedicated locations.
- For small applications, consider housing all DB API functions in a single module, which becomes the DB API layer.
- In larger applications, dividing the DB API layer into multiple modules is advisable. For example, organize user management functions in `db_api/users.py` and order processing functions in `db_api/orders.py`.
- For systems with distinct, independent subsystems, consider placing common functions in a shared location, such as the `db/db_api/` folder, and specific functions in subsystem folders like `external_api/db_api/`.
- Declare the "Db..." dataclasses immediately preceding the functions that produce them.
2. Naming:
- Prefix classes returned from DB API functions with "Db". For instance, use "DbUsers", "DbOrders", "DbOrdersWithDetails", "DbInvoicesReportLine".
- Prefix functions that SELECT data from the DB with "get_". For example, "get_user_by_id", "get_orders_by_user", etc.
- Use prefix "iter_" for DB API functions made using `@nm.sql_iterate` and `@nm.sql_iterate_scalars` decorators.
- For data manipulation functions:
- Use prefixes "ins_", "upd_", "del_" for INSERTs, UPDATEs, DELETEs respectively.
- Employ "upsert_" for upsert operations.
## Advanced features
#### Cancelling operations
If, for any reason, you need to terminate execution and produce a default result in your function, raise the `nm.CancelExecException` exception. Example:
```python
@nm.sql_fetch_all(DbOrder)
def get_orders_by_ids(order_ids: list[int]):
if not order_ids:
raise nm.CancelExecException
return select(Order.id, Order.date, Order.amount).where(
Order.id.in_(order_ids) # <<< empty list is not acceptable here
)
```
The `nm.CancelExecException` triggers production of a default empty result without querying the DB:
- `@nm.sql_fetch_all` and `@nm.sql_fetch_scalars` return an empty list
- `@nm.sql_one_or_none` and `@nm.sql_scalar_or_none` return None
- `@nm.sql_execute` takes no action
#### Registry
Observability is crucial. This library facilitates collecting statistics on DB API function usage out of the box. Statistics include:
- **calls** – number of calls
- **duration** – total execution time
- **tuples** – total number of retrieved tuples
- **fails** – number of failed calls
- **fails_by_error** – dict[str, int] – detailed fails categorized by exception types
```python
from noorm.registry import get_registry
registry = get_registry()
stat = registry.stat_by_name["db.db_api.orders.get_orders_by_user"]
print(stat) # Stat(calls=3, duration=0.0324, tuples=11, fails=0, fails_by_error={})
```
To collect statistics in a multiprocessing application, initialize this option in your MainProcess:
```python
# Example for uvicorn
import uvicorn
from noorm.registry import get_registry
async def app(scope, receive, send):
...
if __name__ == "__main__":
get_registry().init_multiprocess_registry() # <<< here
uvicorn.run("main:app", port=5000, workers=3, log_level="info")
```
Consequently, all DB operations occurring in child processes will be aggregated in the MainProcess registry.
**Important**: statistics for `@nm.sql_iterate` and `@nm.sql_iterate_scalars` is not precise:
1. `stat.duration` is counted only for query execution and first row extraction.
2. `stat.tuples` is always zero.
3. `stat.fails` and `stat.fails_by_error` do not counter errors that might happen after successful first row extraction.
#### Executing unwrapped functions
To call an unwrapped version of a DB API function for evaluation, testing, or debugging purposes:
```python
orders_list = await get_orders_by_user(session, 1) # A "normal" call
orders_list_q = get_orders_by_user.unwrapped(1) # An "unwrapped" call
print(str(orders_list_q.compile())) # Want to see a compiled SqlAlchemy "SELECT ..."
```
## Acknowledgements
Inspired and sponsored by [FRAMEN](https://www.framen.com/)
Raw data
{
"_id": null,
"home_page": "https://github.com/amaslyaev/noorm/",
"name": "true-noorm",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.10",
"maintainer_email": null,
"keywords": "database, sql, orm, noorm, sqlite, postgres, postgresql, mysql",
"author": "Alexander Maslyeav",
"author_email": "maslyaev@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/0f/cb/1b130707e6228d0db9b35b43e8257d4e9b0e89c7d64c4f5e5e134232d5ea/true_noorm-0.1.7.tar.gz",
"platform": null,
"description": "[](https://pypi.python.org/pypi/true-noorm/)\n[](https://pypi.python.org/pypi/true-noorm/)\n[](https://github.com/amaslyaev/noorm/blob/master/LICENSE)\n[](https://codecov.io/gh/amaslyaev/noorm)\n[](https://github.com/psf/black/)\n\n## noorm\n\nNoORM (Not only ORM) - make your database operations convenient and natural\n\n## Install\n\n**noorm** requires Python 3.10 or newer. Install it from PyPI:\n```shell\n$ pip install true-noorm\n```\nPlease note that the correct name is \"**true**-noorm\".\n\n## NoORM principles\n\n1. It is not a holy war against ORM but \"in addition to\".\n2. It is not one more \"finally perfect\" ORM. It is not an ORM at all. **No persistent objects, no \"ideal\" entities.**\n3. It should be good for for medium-sized and big project.\n4. Focus on developer experience.\n5. Not only a set of helpers to write less code, but first of all, an approach that guides to more understandable, performant, scalable, robust, and maintainable solutions.\n\n## Usage\n\nImpotring `noorm` depends on DB you use in your project. Available options:\n\n- `import noorm.sqlite3 as nm` \u2013 for SQLite\n- `import noorm.aiosqlite as nm` \u2013 for asynchronous SQLite via **aiosqlite**\n- `import noorm.psycopg2 as nm` \u2013 for synchronous Postgres via **psycopg2**\n- `import noorm.asyncpg as nm` \u2013 for asynchronous Postgres via **asyncpg**\n- `import noorm.pymysql as nm` \u2013 for synchronous MySQL/MariaDB via **PyMySQL**\n- `import noorm.aiomysql as nm` \u2013 for asynchronous MySQL/MariaDB via **aiomysql**\n- `import noorm.sqlalchemy_sync as nm` \u2013 for synchronous **SqlAlchemy**\n- `import noorm.sqlalchemy_async as nm` \u2013 for asynchronous **SqlAlchemy**\n\nYes, using the SqlAlchemy ORM through NoORM is a nice idea.\n\nAfter importing \"nm\" you use `@nm.sql_...` decorators to create functions that perform database operations. All other your application code uses these functions as a so-called \"DB API layer\". The decorators are:\n- **@nm.sql_fetch_all** \u2013 to make a query and produce a list of objects of specified type. The query is usually SELECT, but it is also useful with data manipulations RETURNING data.\n- **@nm.sql_one_or_none** \u2013 to make a query and produce one object or None if nothing is found.\n- **@nm.sql_scalar_or_none** \u2013 to get a scalar or None if nothing is found.\n- **@nm.sql_fetch_scalars** \u2013 to get a list of scalars.\n- **@nm.sql_execute** \u2013 to execute something, usually INSERT, UPDATE, or DELETE.\n- **@nm.sql_iterate** and **@nm.sql_iterate_scalars** \u2013 to make a query and iterate through results \u2013 objects or scalars respectively. Be careful with this features and, if possible, use `sql_fetch_all` and `sql_fetch_scalars` instead, because they give you less possibilites to shoot your leg. These functions are not implemented for **asyncpg**.\n\nUsage of these decorators in different submodules and underlying databases might have own peculiarities, so check docstring documentation of the chosen \"nm\".\n\n#### Example for SQLite through the sqlite3 standard library\n\nAssume we have a **users** table with fields:\n- Integer `id` (`rowid` in SQLite)\n- String `username`\n- String `email`\n\nAnd an **orders** table:\n- Integer `id` (`rowid` in SQLite)\n- Integer `user_id` references to user id\n- Datetime `order_date` (TEXT in SQLite)\n- Decimal `amount` (TEXT in SQLite)\n\n```python\nfrom dataclasses import dataclass\nfrom decimal import Decimal\nfrom datetime import datetime\nimport sqlite3\nimport noorm.sqlite3 as nm\n\n@dataclass\nclass DbUser: # When we need only basic info. Not a \"model\"! Just a dataclass!\n id: int\n username: str\n email: str\n\n@nm.sql_fetch_all(DbUser, \"SELECT rowid AS id, username, email FROM users\")\ndef get_all_users():\n pass # no parameters, so just \"pass\"\n\n@nm.sql_one_or_none(\n DbUser, \"SELECT rowid AS id, username, email FROM users WHERE rowid = :id\"\n)\ndef get_user_by_id(id_: int):\n return nm.params(id=id_)\n\n@dataclass\nclass DbUserWithOrdersSummary: # With additional info from `orders` table\n id: int\n username: str\n sum_orders: Decimal | None # SQLite noorm can make decimal out of TEXT\n first_order: datetime | None # and datatime too.\n last_order: datetime | None\n\n@nm.sql_fetch_all(\n DbUserWithOrdersSummary,\n \"\"\"SELECT\n u.rowid AS id, u.username,\n SUM(o.amount) AS sum_orders,\n MIN(o.order_date) AS first_order, MAX(o.order_date) AS last_order\n FROM users u\n LEFT OUTER JOIN orders o ON o.user_id = u.rowid\n GROUP BY u.rowid, u.username ORDER BY u.rowid\n \"\"\",\n)\ndef get_users_with_order_summary():\n pass\n\ndef main():\n with sqlite3.connect(\"data.sqlite\") as conn:\n # will use our DB API functions\n for usr in get_all_users(conn):\n print(usr)\n\n print(f\"{get_user_by_id(conn, 1)=}\")\n\n for usr_summary in get_users_with_order_summary(conn):\n print(usr_summary)\n```\n\n#### Example for SQLite through SqlAlchemy\n\nWill use asynchronous version of SqlAlchemy \"nm\".\n```python\nimport sqlalchemy as sa\nfrom sqlalchemy.ext.asyncio import create_async_engine, AsyncSession\nimport noorm.sqlalchemy_async as nm\n...\n\n@nm.sql_fetch_all(DbUser)\ndef get_all_users(): # Notice no \"async\"\n return sa.select(User.id, User.username, User.email)\n\n@nm.sql_one_or_none(DbUser)\ndef get_user_by_id(id_: int):\n return sa.select(User.id, User.username, User.email).where(User.id == id_)\n\n@nm.sql_fetch_all(DbUserWithOrdersSummary)\ndef get_users_with_order_summary():\n return (\n sa.select(\n User.id,\n User.username,\n sa.func.sum(Order.amount).label(\"sum_orders\"),\n sa.func.min(Order.order_date).label(\"first_order\"),\n sa.func.max(Order.order_date).label(\"last_order\"),\n )\n .select_from(User) # `User` and `Order` are ORM model classes\n .outerjoin(Order, Order.user_id == User.id)\n .group_by(User.id)\n .order_by(User.id)\n )\n\nasync def main():\n engine = create_async_engine(\"sqlite+aiosqlite:///data.sqlite\")\n async with AsyncSession(engine) as session:\n for usr in await get_all_users(session): # Notice \"await\"\n print(usr)\n\n print(f\"user_1={await get_user_by_id(session, 1)}\")\n\n for usr_summary in await get_users_with_order_summary(session):\n print(usr_summary)\n```\n\n## Suggestions\n\n1. Code structure:\n - Avoid scattering DB API functions throughout the codebase. It's preferable to consolidate them in dedicated locations.\n - For small applications, consider housing all DB API functions in a single module, which becomes the DB API layer.\n - In larger applications, dividing the DB API layer into multiple modules is advisable. For example, organize user management functions in `db_api/users.py` and order processing functions in `db_api/orders.py`.\n - For systems with distinct, independent subsystems, consider placing common functions in a shared location, such as the `db/db_api/` folder, and specific functions in subsystem folders like `external_api/db_api/`.\n - Declare the \"Db...\" dataclasses immediately preceding the functions that produce them.\n2. Naming:\n - Prefix classes returned from DB API functions with \"Db\". For instance, use \"DbUsers\", \"DbOrders\", \"DbOrdersWithDetails\", \"DbInvoicesReportLine\".\n - Prefix functions that SELECT data from the DB with \"get_\". For example, \"get_user_by_id\", \"get_orders_by_user\", etc.\n - Use prefix \"iter_\" for DB API functions made using `@nm.sql_iterate` and `@nm.sql_iterate_scalars` decorators.\n - For data manipulation functions:\n - Use prefixes \"ins_\", \"upd_\", \"del_\" for INSERTs, UPDATEs, DELETEs respectively.\n - Employ \"upsert_\" for upsert operations.\n\n## Advanced features\n\n#### Cancelling operations\n\nIf, for any reason, you need to terminate execution and produce a default result in your function, raise the `nm.CancelExecException` exception. Example:\n```python\n@nm.sql_fetch_all(DbOrder)\ndef get_orders_by_ids(order_ids: list[int]):\n if not order_ids:\n raise nm.CancelExecException\n return select(Order.id, Order.date, Order.amount).where(\n Order.id.in_(order_ids) # <<< empty list is not acceptable here\n )\n```\nThe `nm.CancelExecException` triggers production of a default empty result without querying the DB:\n- `@nm.sql_fetch_all` and `@nm.sql_fetch_scalars` return an empty list\n- `@nm.sql_one_or_none` and `@nm.sql_scalar_or_none` return None\n- `@nm.sql_execute` takes no action\n\n#### Registry\n\nObservability is crucial. This library facilitates collecting statistics on DB API function usage out of the box. Statistics include:\n- **calls** \u2013 number of calls\n- **duration** \u2013 total execution time\n- **tuples** \u2013 total number of retrieved tuples\n- **fails** \u2013 number of failed calls\n- **fails_by_error** \u2013 dict[str, int] \u2013 detailed fails categorized by exception types\n```python\nfrom noorm.registry import get_registry\nregistry = get_registry()\nstat = registry.stat_by_name[\"db.db_api.orders.get_orders_by_user\"]\nprint(stat) # Stat(calls=3, duration=0.0324, tuples=11, fails=0, fails_by_error={})\n```\nTo collect statistics in a multiprocessing application, initialize this option in your MainProcess:\n```python\n# Example for uvicorn\nimport uvicorn\nfrom noorm.registry import get_registry\n\nasync def app(scope, receive, send):\n ...\n\nif __name__ == \"__main__\":\n get_registry().init_multiprocess_registry() # <<< here\n uvicorn.run(\"main:app\", port=5000, workers=3, log_level=\"info\")\n```\nConsequently, all DB operations occurring in child processes will be aggregated in the MainProcess registry.\n\n**Important**: statistics for `@nm.sql_iterate` and `@nm.sql_iterate_scalars` is not precise:\n1. `stat.duration` is counted only for query execution and first row extraction.\n2. `stat.tuples` is always zero.\n3. `stat.fails` and `stat.fails_by_error` do not counter errors that might happen after successful first row extraction.\n\n#### Executing unwrapped functions\n\nTo call an unwrapped version of a DB API function for evaluation, testing, or debugging purposes:\n```python\norders_list = await get_orders_by_user(session, 1) # A \"normal\" call\norders_list_q = get_orders_by_user.unwrapped(1) # An \"unwrapped\" call\nprint(str(orders_list_q.compile())) # Want to see a compiled SqlAlchemy \"SELECT ...\"\n```\n\n## Acknowledgements\n\nInspired and sponsored by [FRAMEN](https://www.framen.com/)\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "NoORM (Not only ORM) - make your database operations convenient and natural",
"version": "0.1.7",
"project_urls": {
"Homepage": "https://github.com/amaslyaev/noorm/",
"Repository": "https://github.com/amaslyaev/noorm/"
},
"split_keywords": [
"database",
" sql",
" orm",
" noorm",
" sqlite",
" postgres",
" postgresql",
" mysql"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "e69d94db2744afe27af564f8d6be1d7ff467edf90b9cca3a5f56a7da0a2f711d",
"md5": "c3d881b5849d23052382143cfe968185",
"sha256": "90f952c7725e22aa4037efbca9a30bcb807792b6c54341a724059f0ab11475d3"
},
"downloads": -1,
"filename": "true_noorm-0.1.7-py3-none-any.whl",
"has_sig": false,
"md5_digest": "c3d881b5849d23052382143cfe968185",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.10",
"size": 42604,
"upload_time": "2024-09-11T17:11:32",
"upload_time_iso_8601": "2024-09-11T17:11:32.048323Z",
"url": "https://files.pythonhosted.org/packages/e6/9d/94db2744afe27af564f8d6be1d7ff467edf90b9cca3a5f56a7da0a2f711d/true_noorm-0.1.7-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "0fcb1b130707e6228d0db9b35b43e8257d4e9b0e89c7d64c4f5e5e134232d5ea",
"md5": "33c6a5b899bb21bb52912ea983e38838",
"sha256": "fffb038666903319c52635fbbc2dace9b90fda479d782c6eb32c78c0eb0e359c"
},
"downloads": -1,
"filename": "true_noorm-0.1.7.tar.gz",
"has_sig": false,
"md5_digest": "33c6a5b899bb21bb52912ea983e38838",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.10",
"size": 27163,
"upload_time": "2024-09-11T17:11:33",
"upload_time_iso_8601": "2024-09-11T17:11:33.952661Z",
"url": "https://files.pythonhosted.org/packages/0f/cb/1b130707e6228d0db9b35b43e8257d4e9b0e89c7d64c4f5e5e134232d5ea/true_noorm-0.1.7.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-11 17:11:33",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "amaslyaev",
"github_project": "noorm",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "true-noorm"
}
JSON
