diff options
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | python-pynocular.spec | 1513 | ||||
| -rw-r--r-- | sources | 1 |
3 files changed, 1515 insertions, 0 deletions
@@ -0,0 +1 @@ +/pynocular-0.19.0.tar.gz diff --git a/python-pynocular.spec b/python-pynocular.spec new file mode 100644 index 0000000..38c1154 --- /dev/null +++ b/python-pynocular.spec @@ -0,0 +1,1513 @@ +%global _empty_manifest_terminate_build 0 +Name: python-pynocular +Version: 0.19.0 +Release: 1 +Summary: Lightweight ORM that lets you query your database using Pydantic models and asyncio +License: BSD-3-Clause +URL: https://github.com/NarrativeScience/pynocular +Source0: https://mirrors.nju.edu.cn/pypi/web/packages/13/72/769013d6f9e9d50527881aec1dde3b22afd08e5fef6151c76b27fd85eba1/pynocular-0.19.0.tar.gz +BuildArch: noarch + +Requires: python3-aenum +Requires: python3-aiocontextvars +Requires: python3-aiopg[sa] +Requires: python3-pydantic + +%description +# Pynocular + +[](https://pypi.org/pypi/pynocular/) [](https://opensource.org/licenses/BSD-3-Clause) + +Pynocular is a lightweight ORM that lets you query your database using Pydantic models and asyncio. + +With Pynocular you can decorate your existing Pydantic models to sync them with the corresponding table in your +database, allowing you to persist changes without ever having to think about the database. Transaction management is +automatically handled for you so you can focus on the important parts of your code. This integrates seamlessly with frameworks that use Pydantic models such as FastAPI. + +Features: + +- Fully supports asyncio to write to SQL databases +- Provides simple methods for basic SQLAlchemy support (create, delete, update, read) +- Contains access to more advanced functionality such as custom SQLAlchemy selects +- Contains helper functions for creating new database tables +- Advanced transaction management system allows you to conditionally put requests in transactions + +Table of Contents: + +- [Installation](#installation) +- [Guide](#guide) + - [Basic Usage](#basic-usage) + - [Advanced Usage](#advanced-usage) + - [Creating database tables](#creating-database-tables) +- [Development](#development) + +## Installation + +Pynocular requires Python 3.6 or above. + +```bash +pip install pynocular +# or +poetry add pynocular +``` + +## Guide + +### Basic Usage + +Pynocular works by decorating your base Pydantic model with the function `database_model`. Once decorated +with the proper information, you can proceed to use that model to interface with your specified database table. + +The first step is to define a `DBInfo` object. This will contain the connection information to your database. + +```python +from pynocular.engines import DatabaseType, DBInfo + + +# Example below shows how to connect to a locally-running Postgres database +connection_string = f"postgresql://{db_user_name}:{db_user_password}@localhost:5432/{db_name}?sslmode=disable" +) +db_info = DBInfo(connection_string) +``` + +#### Object Management + +Once you define a `db_info` object, you are ready to decorate your Pydantic models and interact with your database! + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, UUID_STR + +from my_package import db_info + +@database_model("organizations", db_info) +class Org(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + name: str = Field(max_length=45) + slug: str = Field(max_length=45) + tag: Optional[str] = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + +#### Object management + +# Create a new Org via `create` +org = await Org.create(name="new org", slug="new-org") + + +# Create a new Org via `save` +org2 = Org(name="new org2", slug="new-org2") +await org2.save() + + +# Update an org +org.name = "renamed org" +await org.save() + + +# Delete org +await org.delete() + + +# Get org +org3 = await Org.get(org2.id) +assert org3 == org2 + +# Get a list of orgs +orgs = await Org.get_list() + +# Get a filtered list of orgs +orgs = await Org.get_list(tag="green") + +# Get orgs that have several different tags +orgs = await Org.get_list(tag=["green", "blue", "red"]) + +# Fetch the latest state of a table in the db +org3.name = "fake name" +await org3.fetch() +assert org3.name == "new org2" + +``` + +#### Serialization + +DatabaseModels have their own serialization functions to convert to and from +dictionaries. + +```python +# Serializing org with `to_dict()` +org = Org.create(name="org serialize", slug="org-serialize") +org_dict = org.to_dict() +expected_org_dict = { + "id": "e64f6c7a-1bd1-4169-b482-189bd3598079", + "name": "org serialize", + "slug": "org-serialize", + "created_at": "2018-01-01 7:03:45", + "updated_at": "2018-01-01 9:24:12" +} +assert org_dict == expected_org_dict + + +# De-serializing org with `from_dict()` +new_org = Org.from_dict(expected_org_dict) +assert org == new_org +``` + +#### Using Nested DatabaseModels + +Pynocular also supports basic object relationships. If your database tables have a +foreign key reference you can leverage that in your pydantic models to increase the +accessibility of those related objects. + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, nested_model, UUID_STR + +from my_package import db_info + +@database_model("users", db_info) +class User(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + username: str = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + +@database_model("organizations", db_info) +class Org(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + name: str = Field(max_length=45) + slug: str = Field(max_length=45) + # `organizations`.`tech_owner_id` is a foreign key to `users`.`id` + tech_owner: Optional[nested_model(User, reference_field="tech_owner_id")] + # `organizations`.`business_owner_id` is a foreign key to `users`.`id` + business_owner: nested_model(User, reference_field="business_owner_id") + tag: Optional[str] = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + + +tech_owner = await User.create("tech owner") +business_owner = await User.create("business owner") + + +# Creating org with only business owner set +org = await Org.create( + name="org name", + slug="org-slug", + business_owner=business_owner +) + +assert org.business_owner == business_owner + +# Add tech owner +org.tech_owner = tech_owner +await org.save() + +# Fetch from the db and check ids +org2 = Org.get(org.id) +assert org2.tech_owner.id == tech_owner.id +assert org2.business_owner.id == business_owner.id + +# Swap user roles +org2.tech_owner = business_owner +org2.business_owner = tech_owner +await org2.save() +org3 = await Org.get(org2.id) +assert org3.tech_owner.id == business_owner.id +assert org3.business_owner.id == tech_owner.id + + +# Serialize org +org_dict = org3.to_dict() +expected_org_dict = { + "id": org3.id, + "name": "org name", + "slug": "org-slug", + "business_owner_id": tech_owner.id, + "tech_owner_id": business_owner.id, + "tag": None, + "created_at": org3.created_at, + "updated_at": org3.updated_at +} + +assert org_dict == expected_org_dict + +``` + +When using `DatabaseModel.get(..)`, any foreign references will need to be resolved before any properties besides the primary ID can be accessed. If you try to access a property before calling `fetch()` on the nested model, a `NestedDatabaseModelNotResolved` error will be thrown. + +```python +org_get = await Org.get(org3.id) +org_get.tech_owner.id # Does not raise `NestedDatabaseModelNotResolved` +org_get.tech_owner.username # Raises `NestedDatabaseModelNotResolved` + +org_get = await Org.get(org3.id) +await org_get.tech_owner.fetch() +org_get.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved` +``` + +Alternatively, calling `DatabaseModel.get_with_refs()` instead of `DatabaseModel.get()` will +automatically fetch the referenced records and fully resolve those objects for you. + +```python +org_get_with_refs = await Org.get_with_refs(org3.id) +org_get_with_refs.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved` +``` + +There are some situations where none of the objects have been persisted to the +database yet. In this situation, you can call `Database.save(include_nested_models=True)` +on the object with the references and it will persist all of them in a transaction. + +```python +# We create the objects but dont persist them +tech_owner = User("tech owner") +business_owner = User("business owner") + +org = Org( + name="org name", + slug="org-slug", + business_owner=business_owner +) + +await org.save(include_nested_models=True) +``` + +#### Special Type arguments + +With Pynocular you can set fields to be optional and set by the database. This is useful +if you want to let the database autogenerate your primary key or `created_at` and `updated_at` fields +on your table. To do this you must: + +- Wrap the typehint in `Optional` +- Provide keyword arguments of `fetch_on_create=True` or `fetch_on_update=True` to the `Field` class + +### Advanced Usage + +For most use cases, the basic usage defined above should suffice. However, there are certain situations +where you don't necessarily want to fetch each object or you need to do more complex queries that +are not exposed by the `DatabaseModel` interface. Below are some examples of how those situations can +be addressed using Pynocular. + +#### Tables with compound keys + +Pynocular supports tables that use multiple fields as its primary key such as join tables. + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, nested_model, UUID_STR + +from my_package import db_info + +@database_model("user_subscriptions", db_info) +class UserSubscriptions(BaseModel): + + user_id: UUID_STR = Field(primary_key=True, fetch_on_create=True) + subscription_id: UUID_STR = Field(primary_key=True, fetch_on_create=True) + name: str + + +user_sub = await UserSub.create( + user_id="4d4254c4-8e99-45f9-8261-82f87991c659", + subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d", + name="User 1's subscriptions" +) + +# Get the users subscription and confirm its the same +user_sub_get = await UserSub.get( + user_id="4d4254c4-8e99-45f9-8261-82f87991c659", + subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d", +) +assert user_sub_get == user_sub + +# Change a property value like any other object +user_sub_get.name = "change name" +await user_sub_get.save() +``` + +#### Batch operations on tables + +Sometimes you want to insert a bunch of records into a database and you don't want to do an insert for each one. +This can be handled by the `create_list` function. + +```python +org_list = [ + Org(name="org1", slug="org-slug1"), + Org(name="org2", slug="org-slug2"), + Org(name="org3", slug="org-slug3"), +] +await Org.create_list(org_list) +``` + +This function will insert all records into your database table in one batch. + +If you have a use case that requires deleting a bunch of records based on some field value, you can use `delete_records`: + +```python +# Delete all records with the tag "green" +await Org.delete_records(tag="green") + +# Delete all records with if their tag has any of the following: "green", "blue", "red" +await Org.delete_records(tag=["green", "blue", "red"]) +``` + +Sometimes you may want to update the value of a record in a database without having to fetch it first. This can be accomplished by using +the `update_record` function: + +```python +await Org.update_record( + id="05c0060c-ceb8-40f0-8faa-dfb91266a6cf", + tag="blue" +) +org = await Org.get("05c0060c-ceb8-40f0-8faa-dfb91266a6cf") +assert org.tag == "blue" +``` + +#### Complex queries + +Sometimes your application will require performing complex queries, such as getting the count of each unique field value for all records in the table. +Because Pynocular is backed by SQLAlchemy, we can access table columns directly to write pure SQLAlchemy queries as well! + +```python +from sqlalchemy import func, select +from pynocular.engines import DBEngine +async def generate_org_stats(): + query = ( + select([func.count(Org.column.id), Org.column.tag]) + .group_by(Org.column.tag) + .order_by(func.count().desc()) + ) + async with await DBEngine.transaction(Org._database_info, is_conditional=False) as conn: + result = await conn.execute(query) + return [dict(row) async for row in result] +``` + +NOTE: `DBengine.transaction` is used to create a connection to the database using the credentials passed in. +If `is_conditional` is `False`, then it will add the query to any transaction that is opened in the call chain. This allows us to make database calls +in different functions but still have them all be under the same database transaction. If there is no transaction opened in the call chain it will open +a new one and any subsequent calls underneath that context manager will be added to the new transaction. + +If `is_conditional` is `True` and there is no transaction in the call chain, then the connection will not create a new transaction. Instead, the query will be performed without a transaction. + +### Creating database and tables + +With Pynocular you can use simple python code to create new databases and database tables. All you need is a working connection string to the database host, a `DatabaseInfo` object that contains the information of the database you want to create, and a properly decorated pydantic model. When you decorate a pydantic model with Pynocular, it creates a SQLAlchemy table as a private variable. This can be accessed via the `_table` property +(although accessing private variables is not recommended). + +```python +from pynocular.db_util import create_new_database, create_table + +from my_package import Org, db_info + +connection_string = "postgresql://postgres:XXXX@localhost:5432/postgres?sslmode=disable" + +# Creates a new database and "organizations" table in that database +await create_new_database(connection_string, db_info) +await create_table(db_info, Org._table) + +``` + +### Unit Testing with DatabaseModels + +Pynocular comes with tooling to write unit tests against your DatabaseModels, giving you +the ability to test your business logic without the extra work and latency involved in +managing a database. All you have to do is use the `patch_database_model` context +manager provided in Pynocular. + +```python +from pynocular.patch_models import patch_database_model + +from my_package import Org, User + + +with patch_database_model(Org): + orgs = [ + Org(id=str(uuid4()), name="orgus borgus", slug="orgus_borgus"), + Org(id=str(uuid4()), name="orgus borgus2", slug="orgus_borgus"), + ] + + await Org.create_list(orgs) + fetched_orgs = await Org.get_list(name=orgs[0].name) + assert orgs[0] == fetched_orgs[0] + +# patch_database_model also works with nested models +users = [ + User(id=str(uuid4()), username="Bob"), + User(id=str(uuid4()), username="Sally"), +] +orgs = [ + Org( + id=str(uuid4()), + name="orgus borgus", + slug="orgus_borgus", + tech_owner=users[0], + business_owner=users[1], + ), +] + +with patch_database_model(Org, models=orgs), patch_database_model( + User, models=users +): + org = await Org.get(orgs[0].id) + org.name = "new test name" + users[0].username = "bberkley" + + # Save the username update when saving the org model update + await org.save(include_nested_models=True) + + # Get the org with the resolved nested model + org_get = await Org.get_with_refs(org.id) + assert org_get.name == "new test name" + assert org_get.tech_owner.username == "bberkley" +``` + +## Development + +To develop Pynocular, install dependencies and enable the pre-commit hook. + +The example below is using Python 3.9 but you can replace this with any supported version of Python. + +Install Python 3.9 and activate it in your shell. + +```bash +sudo yum install libffi-devel # Needed for ctypes to install poetry +pyenv install 3.9.7 +pyenv shell 3.9.7 +``` + +Install dependencies and enable the pre-commit hook. + +```bash +pip install pre-commit poetry +poetry install +pre-commit install +``` + +Run tests to confirm everything is installed correctly. + +```bash +poetry run pytest +``` + + +%package -n python3-pynocular +Summary: Lightweight ORM that lets you query your database using Pydantic models and asyncio +Provides: python-pynocular +BuildRequires: python3-devel +BuildRequires: python3-setuptools +BuildRequires: python3-pip +%description -n python3-pynocular +# Pynocular + +[](https://pypi.org/pypi/pynocular/) [](https://opensource.org/licenses/BSD-3-Clause) + +Pynocular is a lightweight ORM that lets you query your database using Pydantic models and asyncio. + +With Pynocular you can decorate your existing Pydantic models to sync them with the corresponding table in your +database, allowing you to persist changes without ever having to think about the database. Transaction management is +automatically handled for you so you can focus on the important parts of your code. This integrates seamlessly with frameworks that use Pydantic models such as FastAPI. + +Features: + +- Fully supports asyncio to write to SQL databases +- Provides simple methods for basic SQLAlchemy support (create, delete, update, read) +- Contains access to more advanced functionality such as custom SQLAlchemy selects +- Contains helper functions for creating new database tables +- Advanced transaction management system allows you to conditionally put requests in transactions + +Table of Contents: + +- [Installation](#installation) +- [Guide](#guide) + - [Basic Usage](#basic-usage) + - [Advanced Usage](#advanced-usage) + - [Creating database tables](#creating-database-tables) +- [Development](#development) + +## Installation + +Pynocular requires Python 3.6 or above. + +```bash +pip install pynocular +# or +poetry add pynocular +``` + +## Guide + +### Basic Usage + +Pynocular works by decorating your base Pydantic model with the function `database_model`. Once decorated +with the proper information, you can proceed to use that model to interface with your specified database table. + +The first step is to define a `DBInfo` object. This will contain the connection information to your database. + +```python +from pynocular.engines import DatabaseType, DBInfo + + +# Example below shows how to connect to a locally-running Postgres database +connection_string = f"postgresql://{db_user_name}:{db_user_password}@localhost:5432/{db_name}?sslmode=disable" +) +db_info = DBInfo(connection_string) +``` + +#### Object Management + +Once you define a `db_info` object, you are ready to decorate your Pydantic models and interact with your database! + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, UUID_STR + +from my_package import db_info + +@database_model("organizations", db_info) +class Org(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + name: str = Field(max_length=45) + slug: str = Field(max_length=45) + tag: Optional[str] = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + +#### Object management + +# Create a new Org via `create` +org = await Org.create(name="new org", slug="new-org") + + +# Create a new Org via `save` +org2 = Org(name="new org2", slug="new-org2") +await org2.save() + + +# Update an org +org.name = "renamed org" +await org.save() + + +# Delete org +await org.delete() + + +# Get org +org3 = await Org.get(org2.id) +assert org3 == org2 + +# Get a list of orgs +orgs = await Org.get_list() + +# Get a filtered list of orgs +orgs = await Org.get_list(tag="green") + +# Get orgs that have several different tags +orgs = await Org.get_list(tag=["green", "blue", "red"]) + +# Fetch the latest state of a table in the db +org3.name = "fake name" +await org3.fetch() +assert org3.name == "new org2" + +``` + +#### Serialization + +DatabaseModels have their own serialization functions to convert to and from +dictionaries. + +```python +# Serializing org with `to_dict()` +org = Org.create(name="org serialize", slug="org-serialize") +org_dict = org.to_dict() +expected_org_dict = { + "id": "e64f6c7a-1bd1-4169-b482-189bd3598079", + "name": "org serialize", + "slug": "org-serialize", + "created_at": "2018-01-01 7:03:45", + "updated_at": "2018-01-01 9:24:12" +} +assert org_dict == expected_org_dict + + +# De-serializing org with `from_dict()` +new_org = Org.from_dict(expected_org_dict) +assert org == new_org +``` + +#### Using Nested DatabaseModels + +Pynocular also supports basic object relationships. If your database tables have a +foreign key reference you can leverage that in your pydantic models to increase the +accessibility of those related objects. + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, nested_model, UUID_STR + +from my_package import db_info + +@database_model("users", db_info) +class User(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + username: str = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + +@database_model("organizations", db_info) +class Org(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + name: str = Field(max_length=45) + slug: str = Field(max_length=45) + # `organizations`.`tech_owner_id` is a foreign key to `users`.`id` + tech_owner: Optional[nested_model(User, reference_field="tech_owner_id")] + # `organizations`.`business_owner_id` is a foreign key to `users`.`id` + business_owner: nested_model(User, reference_field="business_owner_id") + tag: Optional[str] = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + + +tech_owner = await User.create("tech owner") +business_owner = await User.create("business owner") + + +# Creating org with only business owner set +org = await Org.create( + name="org name", + slug="org-slug", + business_owner=business_owner +) + +assert org.business_owner == business_owner + +# Add tech owner +org.tech_owner = tech_owner +await org.save() + +# Fetch from the db and check ids +org2 = Org.get(org.id) +assert org2.tech_owner.id == tech_owner.id +assert org2.business_owner.id == business_owner.id + +# Swap user roles +org2.tech_owner = business_owner +org2.business_owner = tech_owner +await org2.save() +org3 = await Org.get(org2.id) +assert org3.tech_owner.id == business_owner.id +assert org3.business_owner.id == tech_owner.id + + +# Serialize org +org_dict = org3.to_dict() +expected_org_dict = { + "id": org3.id, + "name": "org name", + "slug": "org-slug", + "business_owner_id": tech_owner.id, + "tech_owner_id": business_owner.id, + "tag": None, + "created_at": org3.created_at, + "updated_at": org3.updated_at +} + +assert org_dict == expected_org_dict + +``` + +When using `DatabaseModel.get(..)`, any foreign references will need to be resolved before any properties besides the primary ID can be accessed. If you try to access a property before calling `fetch()` on the nested model, a `NestedDatabaseModelNotResolved` error will be thrown. + +```python +org_get = await Org.get(org3.id) +org_get.tech_owner.id # Does not raise `NestedDatabaseModelNotResolved` +org_get.tech_owner.username # Raises `NestedDatabaseModelNotResolved` + +org_get = await Org.get(org3.id) +await org_get.tech_owner.fetch() +org_get.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved` +``` + +Alternatively, calling `DatabaseModel.get_with_refs()` instead of `DatabaseModel.get()` will +automatically fetch the referenced records and fully resolve those objects for you. + +```python +org_get_with_refs = await Org.get_with_refs(org3.id) +org_get_with_refs.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved` +``` + +There are some situations where none of the objects have been persisted to the +database yet. In this situation, you can call `Database.save(include_nested_models=True)` +on the object with the references and it will persist all of them in a transaction. + +```python +# We create the objects but dont persist them +tech_owner = User("tech owner") +business_owner = User("business owner") + +org = Org( + name="org name", + slug="org-slug", + business_owner=business_owner +) + +await org.save(include_nested_models=True) +``` + +#### Special Type arguments + +With Pynocular you can set fields to be optional and set by the database. This is useful +if you want to let the database autogenerate your primary key or `created_at` and `updated_at` fields +on your table. To do this you must: + +- Wrap the typehint in `Optional` +- Provide keyword arguments of `fetch_on_create=True` or `fetch_on_update=True` to the `Field` class + +### Advanced Usage + +For most use cases, the basic usage defined above should suffice. However, there are certain situations +where you don't necessarily want to fetch each object or you need to do more complex queries that +are not exposed by the `DatabaseModel` interface. Below are some examples of how those situations can +be addressed using Pynocular. + +#### Tables with compound keys + +Pynocular supports tables that use multiple fields as its primary key such as join tables. + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, nested_model, UUID_STR + +from my_package import db_info + +@database_model("user_subscriptions", db_info) +class UserSubscriptions(BaseModel): + + user_id: UUID_STR = Field(primary_key=True, fetch_on_create=True) + subscription_id: UUID_STR = Field(primary_key=True, fetch_on_create=True) + name: str + + +user_sub = await UserSub.create( + user_id="4d4254c4-8e99-45f9-8261-82f87991c659", + subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d", + name="User 1's subscriptions" +) + +# Get the users subscription and confirm its the same +user_sub_get = await UserSub.get( + user_id="4d4254c4-8e99-45f9-8261-82f87991c659", + subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d", +) +assert user_sub_get == user_sub + +# Change a property value like any other object +user_sub_get.name = "change name" +await user_sub_get.save() +``` + +#### Batch operations on tables + +Sometimes you want to insert a bunch of records into a database and you don't want to do an insert for each one. +This can be handled by the `create_list` function. + +```python +org_list = [ + Org(name="org1", slug="org-slug1"), + Org(name="org2", slug="org-slug2"), + Org(name="org3", slug="org-slug3"), +] +await Org.create_list(org_list) +``` + +This function will insert all records into your database table in one batch. + +If you have a use case that requires deleting a bunch of records based on some field value, you can use `delete_records`: + +```python +# Delete all records with the tag "green" +await Org.delete_records(tag="green") + +# Delete all records with if their tag has any of the following: "green", "blue", "red" +await Org.delete_records(tag=["green", "blue", "red"]) +``` + +Sometimes you may want to update the value of a record in a database without having to fetch it first. This can be accomplished by using +the `update_record` function: + +```python +await Org.update_record( + id="05c0060c-ceb8-40f0-8faa-dfb91266a6cf", + tag="blue" +) +org = await Org.get("05c0060c-ceb8-40f0-8faa-dfb91266a6cf") +assert org.tag == "blue" +``` + +#### Complex queries + +Sometimes your application will require performing complex queries, such as getting the count of each unique field value for all records in the table. +Because Pynocular is backed by SQLAlchemy, we can access table columns directly to write pure SQLAlchemy queries as well! + +```python +from sqlalchemy import func, select +from pynocular.engines import DBEngine +async def generate_org_stats(): + query = ( + select([func.count(Org.column.id), Org.column.tag]) + .group_by(Org.column.tag) + .order_by(func.count().desc()) + ) + async with await DBEngine.transaction(Org._database_info, is_conditional=False) as conn: + result = await conn.execute(query) + return [dict(row) async for row in result] +``` + +NOTE: `DBengine.transaction` is used to create a connection to the database using the credentials passed in. +If `is_conditional` is `False`, then it will add the query to any transaction that is opened in the call chain. This allows us to make database calls +in different functions but still have them all be under the same database transaction. If there is no transaction opened in the call chain it will open +a new one and any subsequent calls underneath that context manager will be added to the new transaction. + +If `is_conditional` is `True` and there is no transaction in the call chain, then the connection will not create a new transaction. Instead, the query will be performed without a transaction. + +### Creating database and tables + +With Pynocular you can use simple python code to create new databases and database tables. All you need is a working connection string to the database host, a `DatabaseInfo` object that contains the information of the database you want to create, and a properly decorated pydantic model. When you decorate a pydantic model with Pynocular, it creates a SQLAlchemy table as a private variable. This can be accessed via the `_table` property +(although accessing private variables is not recommended). + +```python +from pynocular.db_util import create_new_database, create_table + +from my_package import Org, db_info + +connection_string = "postgresql://postgres:XXXX@localhost:5432/postgres?sslmode=disable" + +# Creates a new database and "organizations" table in that database +await create_new_database(connection_string, db_info) +await create_table(db_info, Org._table) + +``` + +### Unit Testing with DatabaseModels + +Pynocular comes with tooling to write unit tests against your DatabaseModels, giving you +the ability to test your business logic without the extra work and latency involved in +managing a database. All you have to do is use the `patch_database_model` context +manager provided in Pynocular. + +```python +from pynocular.patch_models import patch_database_model + +from my_package import Org, User + + +with patch_database_model(Org): + orgs = [ + Org(id=str(uuid4()), name="orgus borgus", slug="orgus_borgus"), + Org(id=str(uuid4()), name="orgus borgus2", slug="orgus_borgus"), + ] + + await Org.create_list(orgs) + fetched_orgs = await Org.get_list(name=orgs[0].name) + assert orgs[0] == fetched_orgs[0] + +# patch_database_model also works with nested models +users = [ + User(id=str(uuid4()), username="Bob"), + User(id=str(uuid4()), username="Sally"), +] +orgs = [ + Org( + id=str(uuid4()), + name="orgus borgus", + slug="orgus_borgus", + tech_owner=users[0], + business_owner=users[1], + ), +] + +with patch_database_model(Org, models=orgs), patch_database_model( + User, models=users +): + org = await Org.get(orgs[0].id) + org.name = "new test name" + users[0].username = "bberkley" + + # Save the username update when saving the org model update + await org.save(include_nested_models=True) + + # Get the org with the resolved nested model + org_get = await Org.get_with_refs(org.id) + assert org_get.name == "new test name" + assert org_get.tech_owner.username == "bberkley" +``` + +## Development + +To develop Pynocular, install dependencies and enable the pre-commit hook. + +The example below is using Python 3.9 but you can replace this with any supported version of Python. + +Install Python 3.9 and activate it in your shell. + +```bash +sudo yum install libffi-devel # Needed for ctypes to install poetry +pyenv install 3.9.7 +pyenv shell 3.9.7 +``` + +Install dependencies and enable the pre-commit hook. + +```bash +pip install pre-commit poetry +poetry install +pre-commit install +``` + +Run tests to confirm everything is installed correctly. + +```bash +poetry run pytest +``` + + +%package help +Summary: Development documents and examples for pynocular +Provides: python3-pynocular-doc +%description help +# Pynocular + +[](https://pypi.org/pypi/pynocular/) [](https://opensource.org/licenses/BSD-3-Clause) + +Pynocular is a lightweight ORM that lets you query your database using Pydantic models and asyncio. + +With Pynocular you can decorate your existing Pydantic models to sync them with the corresponding table in your +database, allowing you to persist changes without ever having to think about the database. Transaction management is +automatically handled for you so you can focus on the important parts of your code. This integrates seamlessly with frameworks that use Pydantic models such as FastAPI. + +Features: + +- Fully supports asyncio to write to SQL databases +- Provides simple methods for basic SQLAlchemy support (create, delete, update, read) +- Contains access to more advanced functionality such as custom SQLAlchemy selects +- Contains helper functions for creating new database tables +- Advanced transaction management system allows you to conditionally put requests in transactions + +Table of Contents: + +- [Installation](#installation) +- [Guide](#guide) + - [Basic Usage](#basic-usage) + - [Advanced Usage](#advanced-usage) + - [Creating database tables](#creating-database-tables) +- [Development](#development) + +## Installation + +Pynocular requires Python 3.6 or above. + +```bash +pip install pynocular +# or +poetry add pynocular +``` + +## Guide + +### Basic Usage + +Pynocular works by decorating your base Pydantic model with the function `database_model`. Once decorated +with the proper information, you can proceed to use that model to interface with your specified database table. + +The first step is to define a `DBInfo` object. This will contain the connection information to your database. + +```python +from pynocular.engines import DatabaseType, DBInfo + + +# Example below shows how to connect to a locally-running Postgres database +connection_string = f"postgresql://{db_user_name}:{db_user_password}@localhost:5432/{db_name}?sslmode=disable" +) +db_info = DBInfo(connection_string) +``` + +#### Object Management + +Once you define a `db_info` object, you are ready to decorate your Pydantic models and interact with your database! + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, UUID_STR + +from my_package import db_info + +@database_model("organizations", db_info) +class Org(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + name: str = Field(max_length=45) + slug: str = Field(max_length=45) + tag: Optional[str] = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + +#### Object management + +# Create a new Org via `create` +org = await Org.create(name="new org", slug="new-org") + + +# Create a new Org via `save` +org2 = Org(name="new org2", slug="new-org2") +await org2.save() + + +# Update an org +org.name = "renamed org" +await org.save() + + +# Delete org +await org.delete() + + +# Get org +org3 = await Org.get(org2.id) +assert org3 == org2 + +# Get a list of orgs +orgs = await Org.get_list() + +# Get a filtered list of orgs +orgs = await Org.get_list(tag="green") + +# Get orgs that have several different tags +orgs = await Org.get_list(tag=["green", "blue", "red"]) + +# Fetch the latest state of a table in the db +org3.name = "fake name" +await org3.fetch() +assert org3.name == "new org2" + +``` + +#### Serialization + +DatabaseModels have their own serialization functions to convert to and from +dictionaries. + +```python +# Serializing org with `to_dict()` +org = Org.create(name="org serialize", slug="org-serialize") +org_dict = org.to_dict() +expected_org_dict = { + "id": "e64f6c7a-1bd1-4169-b482-189bd3598079", + "name": "org serialize", + "slug": "org-serialize", + "created_at": "2018-01-01 7:03:45", + "updated_at": "2018-01-01 9:24:12" +} +assert org_dict == expected_org_dict + + +# De-serializing org with `from_dict()` +new_org = Org.from_dict(expected_org_dict) +assert org == new_org +``` + +#### Using Nested DatabaseModels + +Pynocular also supports basic object relationships. If your database tables have a +foreign key reference you can leverage that in your pydantic models to increase the +accessibility of those related objects. + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, nested_model, UUID_STR + +from my_package import db_info + +@database_model("users", db_info) +class User(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + username: str = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + +@database_model("organizations", db_info) +class Org(BaseModel): + + id: Optional[UUID_STR] = Field(primary_key=True, fetch_on_create=True) + name: str = Field(max_length=45) + slug: str = Field(max_length=45) + # `organizations`.`tech_owner_id` is a foreign key to `users`.`id` + tech_owner: Optional[nested_model(User, reference_field="tech_owner_id")] + # `organizations`.`business_owner_id` is a foreign key to `users`.`id` + business_owner: nested_model(User, reference_field="business_owner_id") + tag: Optional[str] = Field(max_length=100) + + created_at: Optional[datetime] = Field(fetch_on_create=True) + updated_at: Optional[datetime] = Field(fetch_on_update=True) + + +tech_owner = await User.create("tech owner") +business_owner = await User.create("business owner") + + +# Creating org with only business owner set +org = await Org.create( + name="org name", + slug="org-slug", + business_owner=business_owner +) + +assert org.business_owner == business_owner + +# Add tech owner +org.tech_owner = tech_owner +await org.save() + +# Fetch from the db and check ids +org2 = Org.get(org.id) +assert org2.tech_owner.id == tech_owner.id +assert org2.business_owner.id == business_owner.id + +# Swap user roles +org2.tech_owner = business_owner +org2.business_owner = tech_owner +await org2.save() +org3 = await Org.get(org2.id) +assert org3.tech_owner.id == business_owner.id +assert org3.business_owner.id == tech_owner.id + + +# Serialize org +org_dict = org3.to_dict() +expected_org_dict = { + "id": org3.id, + "name": "org name", + "slug": "org-slug", + "business_owner_id": tech_owner.id, + "tech_owner_id": business_owner.id, + "tag": None, + "created_at": org3.created_at, + "updated_at": org3.updated_at +} + +assert org_dict == expected_org_dict + +``` + +When using `DatabaseModel.get(..)`, any foreign references will need to be resolved before any properties besides the primary ID can be accessed. If you try to access a property before calling `fetch()` on the nested model, a `NestedDatabaseModelNotResolved` error will be thrown. + +```python +org_get = await Org.get(org3.id) +org_get.tech_owner.id # Does not raise `NestedDatabaseModelNotResolved` +org_get.tech_owner.username # Raises `NestedDatabaseModelNotResolved` + +org_get = await Org.get(org3.id) +await org_get.tech_owner.fetch() +org_get.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved` +``` + +Alternatively, calling `DatabaseModel.get_with_refs()` instead of `DatabaseModel.get()` will +automatically fetch the referenced records and fully resolve those objects for you. + +```python +org_get_with_refs = await Org.get_with_refs(org3.id) +org_get_with_refs.tech_owner.username # Does not raise `NestedDatabaseModelNotResolved` +``` + +There are some situations where none of the objects have been persisted to the +database yet. In this situation, you can call `Database.save(include_nested_models=True)` +on the object with the references and it will persist all of them in a transaction. + +```python +# We create the objects but dont persist them +tech_owner = User("tech owner") +business_owner = User("business owner") + +org = Org( + name="org name", + slug="org-slug", + business_owner=business_owner +) + +await org.save(include_nested_models=True) +``` + +#### Special Type arguments + +With Pynocular you can set fields to be optional and set by the database. This is useful +if you want to let the database autogenerate your primary key or `created_at` and `updated_at` fields +on your table. To do this you must: + +- Wrap the typehint in `Optional` +- Provide keyword arguments of `fetch_on_create=True` or `fetch_on_update=True` to the `Field` class + +### Advanced Usage + +For most use cases, the basic usage defined above should suffice. However, there are certain situations +where you don't necessarily want to fetch each object or you need to do more complex queries that +are not exposed by the `DatabaseModel` interface. Below are some examples of how those situations can +be addressed using Pynocular. + +#### Tables with compound keys + +Pynocular supports tables that use multiple fields as its primary key such as join tables. + +```python +from pydantic import BaseModel, Field +from pynocular.database_model import database_model, nested_model, UUID_STR + +from my_package import db_info + +@database_model("user_subscriptions", db_info) +class UserSubscriptions(BaseModel): + + user_id: UUID_STR = Field(primary_key=True, fetch_on_create=True) + subscription_id: UUID_STR = Field(primary_key=True, fetch_on_create=True) + name: str + + +user_sub = await UserSub.create( + user_id="4d4254c4-8e99-45f9-8261-82f87991c659", + subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d", + name="User 1's subscriptions" +) + +# Get the users subscription and confirm its the same +user_sub_get = await UserSub.get( + user_id="4d4254c4-8e99-45f9-8261-82f87991c659", + subscription_id="3cc5d476-dbe6-4cc1-9390-49ebd7593a3d", +) +assert user_sub_get == user_sub + +# Change a property value like any other object +user_sub_get.name = "change name" +await user_sub_get.save() +``` + +#### Batch operations on tables + +Sometimes you want to insert a bunch of records into a database and you don't want to do an insert for each one. +This can be handled by the `create_list` function. + +```python +org_list = [ + Org(name="org1", slug="org-slug1"), + Org(name="org2", slug="org-slug2"), + Org(name="org3", slug="org-slug3"), +] +await Org.create_list(org_list) +``` + +This function will insert all records into your database table in one batch. + +If you have a use case that requires deleting a bunch of records based on some field value, you can use `delete_records`: + +```python +# Delete all records with the tag "green" +await Org.delete_records(tag="green") + +# Delete all records with if their tag has any of the following: "green", "blue", "red" +await Org.delete_records(tag=["green", "blue", "red"]) +``` + +Sometimes you may want to update the value of a record in a database without having to fetch it first. This can be accomplished by using +the `update_record` function: + +```python +await Org.update_record( + id="05c0060c-ceb8-40f0-8faa-dfb91266a6cf", + tag="blue" +) +org = await Org.get("05c0060c-ceb8-40f0-8faa-dfb91266a6cf") +assert org.tag == "blue" +``` + +#### Complex queries + +Sometimes your application will require performing complex queries, such as getting the count of each unique field value for all records in the table. +Because Pynocular is backed by SQLAlchemy, we can access table columns directly to write pure SQLAlchemy queries as well! + +```python +from sqlalchemy import func, select +from pynocular.engines import DBEngine +async def generate_org_stats(): + query = ( + select([func.count(Org.column.id), Org.column.tag]) + .group_by(Org.column.tag) + .order_by(func.count().desc()) + ) + async with await DBEngine.transaction(Org._database_info, is_conditional=False) as conn: + result = await conn.execute(query) + return [dict(row) async for row in result] +``` + +NOTE: `DBengine.transaction` is used to create a connection to the database using the credentials passed in. +If `is_conditional` is `False`, then it will add the query to any transaction that is opened in the call chain. This allows us to make database calls +in different functions but still have them all be under the same database transaction. If there is no transaction opened in the call chain it will open +a new one and any subsequent calls underneath that context manager will be added to the new transaction. + +If `is_conditional` is `True` and there is no transaction in the call chain, then the connection will not create a new transaction. Instead, the query will be performed without a transaction. + +### Creating database and tables + +With Pynocular you can use simple python code to create new databases and database tables. All you need is a working connection string to the database host, a `DatabaseInfo` object that contains the information of the database you want to create, and a properly decorated pydantic model. When you decorate a pydantic model with Pynocular, it creates a SQLAlchemy table as a private variable. This can be accessed via the `_table` property +(although accessing private variables is not recommended). + +```python +from pynocular.db_util import create_new_database, create_table + +from my_package import Org, db_info + +connection_string = "postgresql://postgres:XXXX@localhost:5432/postgres?sslmode=disable" + +# Creates a new database and "organizations" table in that database +await create_new_database(connection_string, db_info) +await create_table(db_info, Org._table) + +``` + +### Unit Testing with DatabaseModels + +Pynocular comes with tooling to write unit tests against your DatabaseModels, giving you +the ability to test your business logic without the extra work and latency involved in +managing a database. All you have to do is use the `patch_database_model` context +manager provided in Pynocular. + +```python +from pynocular.patch_models import patch_database_model + +from my_package import Org, User + + +with patch_database_model(Org): + orgs = [ + Org(id=str(uuid4()), name="orgus borgus", slug="orgus_borgus"), + Org(id=str(uuid4()), name="orgus borgus2", slug="orgus_borgus"), + ] + + await Org.create_list(orgs) + fetched_orgs = await Org.get_list(name=orgs[0].name) + assert orgs[0] == fetched_orgs[0] + +# patch_database_model also works with nested models +users = [ + User(id=str(uuid4()), username="Bob"), + User(id=str(uuid4()), username="Sally"), +] +orgs = [ + Org( + id=str(uuid4()), + name="orgus borgus", + slug="orgus_borgus", + tech_owner=users[0], + business_owner=users[1], + ), +] + +with patch_database_model(Org, models=orgs), patch_database_model( + User, models=users +): + org = await Org.get(orgs[0].id) + org.name = "new test name" + users[0].username = "bberkley" + + # Save the username update when saving the org model update + await org.save(include_nested_models=True) + + # Get the org with the resolved nested model + org_get = await Org.get_with_refs(org.id) + assert org_get.name == "new test name" + assert org_get.tech_owner.username == "bberkley" +``` + +## Development + +To develop Pynocular, install dependencies and enable the pre-commit hook. + +The example below is using Python 3.9 but you can replace this with any supported version of Python. + +Install Python 3.9 and activate it in your shell. + +```bash +sudo yum install libffi-devel # Needed for ctypes to install poetry +pyenv install 3.9.7 +pyenv shell 3.9.7 +``` + +Install dependencies and enable the pre-commit hook. + +```bash +pip install pre-commit poetry +poetry install +pre-commit install +``` + +Run tests to confirm everything is installed correctly. + +```bash +poetry run pytest +``` + + +%prep +%autosetup -n pynocular-0.19.0 + +%build +%py3_build + +%install +%py3_install +install -d -m755 %{buildroot}/%{_pkgdocdir} +if [ -d doc ]; then cp -arf doc %{buildroot}/%{_pkgdocdir}; fi +if [ -d docs ]; then cp -arf docs %{buildroot}/%{_pkgdocdir}; fi +if [ -d example ]; then cp -arf example %{buildroot}/%{_pkgdocdir}; fi +if [ -d examples ]; then cp -arf examples %{buildroot}/%{_pkgdocdir}; fi +pushd %{buildroot} +if [ -d usr/lib ]; then + find usr/lib -type f -printf "/%h/%f\n" >> filelist.lst +fi +if [ -d usr/lib64 ]; then + find usr/lib64 -type f -printf "/%h/%f\n" >> filelist.lst +fi +if [ -d usr/bin ]; then + find usr/bin -type f -printf "/%h/%f\n" >> filelist.lst +fi +if [ -d usr/sbin ]; then + find usr/sbin -type f -printf "/%h/%f\n" >> filelist.lst +fi +touch doclist.lst +if [ -d usr/share/man ]; then + find usr/share/man -type f -printf "/%h/%f.gz\n" >> doclist.lst +fi +popd +mv %{buildroot}/filelist.lst . +mv %{buildroot}/doclist.lst . + +%files -n python3-pynocular -f filelist.lst +%dir %{python3_sitelib}/* + +%files help -f doclist.lst +%{_docdir}/* + +%changelog +* Wed May 31 2023 Python_Bot <Python_Bot@openeuler.org> - 0.19.0-1 +- Package Spec generated @@ -0,0 +1 @@ +d2ce88920d46a5fb372ad80224c77369 pynocular-0.19.0.tar.gz |