Welcome to pgcrud
pgcrud is a Python package that makes Create, Read, Update, and Delete (CRUD) operations for PostgreSQL simple and fast.
Key Features
- No ORM, only declarative expressions.
- Built-in pydantic and msgspec support for data serialization & validation
- Efficient handling of complex parent-child relationships
- Perform operations sync or async.
- Full type hint support.
- Easy to integrate into existing projects.
- Protection against SQL-Injection
- Tailored to PostgreSQL with wide extensions support.
Dependencies
pgcrud's only dependency is psycopg including it's binary and pool extensions.
Optional
For data serialization & validation you can choose between:
Installation
pgcrud is not yet available on PyPI. However, you can install it using pip with the following command:
To use pydantic or msgspec, ensure you install them separately, as they are optional dependencies.
Do not download the pgcrud package from PyPi. This is an abandoned package and is not affiliated with us.
Cursor
pgcrud extends the psycopg cursor by directly fetching and validating the data with your preferred data serialization & validation library.
from pydantic import BaseModel
import pgcrud as pg
class Author(BaseModel):
id: int
name: str
with pg.connect('CONN_STR') as conn:
with conn.cursor() as cursor:
cursor[Author].execute("SELECT 1 AS id, 'J.K. Rowling' AS name").fetchone()
# returns Author(id=1, name='J.K. Rowling')
cursor[str].execute("SELECT 1").fetchone()
# raises ValidationError: Input should be a valid string [type=string_type, input_value=1, input_type=int]
import asyncio
from pydantic import BaseModel
import pgcrud as pg
class Author(BaseModel):
id: int
name: str
async def main():
async with await pg.a.connect('CONN_STR') as conn:
async with conn.cursor() as cursor:
await cursor[Author].execute("SELECT 1 AS id, 'J.K. Rowling' AS name")
await cursor.fetchone()
# returns Author(id=1, name='J.K. Rowling')
await cursor[str].execute("SELECT 1")
await cursor.fetchone()
# raises ValidationError: Input should be a valid string [type=string_type, input_value=1, input_type=int]
asyncio.run(main())
CRUD Examples
This is a high level overview of the pgcrud operations. For more detailed instructions and advanced examples, be sure to explore the rest of the documentation.
In the following examples, we use a simple Author model based on an author table with the columns id, name, and date_of_birth.
Create
The insert_one / insert_many methods are used to insert one / many records into a table. You must specify into which table
you want to insert and which columns you want to populate. You can simply insert new records from Pydantic objects. Typically, the id
column is autogenerated and you can retrieve it after insertion.
from datetime import date
from pydantic import BaseModel
import pgcrud as pg
from pgcrud import e
class AuthorInput(BaseModel):
name: str
date_of_birth: date
def insert_author(
cursor: pg.Cursor,
input_: AuthorInput,
) -> int | None:
return pg.insert_one(
cursor=cursor[int],
insert_into=e.author[e.name, e.date_of_birth],
values=input_,
returning=e.id,
)
from datetime import date
from pydantic import BaseModel
import pgcrud as pg
from pgcrud import e
class AuthorInput(BaseModel):
name: str
date_of_birth: date
async def insert_author(
cursor: pg.a.Cursor,
input_: AuthorInput,
) -> int | None:
return await pg.a.insert_one(
cursor=cursor[int],
insert_into=e.author[e.name, e.date_of_birth],
values=input_,
returning=e.id,
)
Read
The get_one / get_many methods are used to retrieve one / many records from a table. You must specify from with table you want to select
the records. You can directly load the fetched records into a Pydantic model. The where condition specifies which records should be retrieved.
If the where condition matches multiple records, the get_one method will return the first record that meets the filter criteria.
from datetime import date
from pydantic import BaseModel
import pgcrud as pg
from pgcrud import e
class Author(BaseModel):
id: int
name: str
date_of_birth: date
def get_author(
cursor: pg.Cursor,
id_: int,
) -> Author | None:
return pg.get_one(
cursor=cursor[Author],
select=(e.id, e.name, e.date_of_birth),
from_=e.author,
where=e.id == id_,
)
from datetime import date
from pydantic import BaseModel
import pgcrud as pg
from pgcrud import e
class Author(BaseModel):
id: int
name: str
date_of_birth: date
async def get_author(
cursor: pg.a.Cursor,
id_: int,
) -> Author | None:
return await pg.a.get_one(
cursor=cursor[Author],
select=(e.id, e.name, e.date_of_birth),
from_=e.author,
where=e.id == id_,
)
Update
The update_many method is used to update records in a table.1 You need to specify which table and which columns you want to update.
The where condition specifies which records should be updated. To update a single record, you must ensure that the where condition
targets a single record.
from datetime import date
from pydantic import BaseModel
import pgcrud as pg
from pgcrud import e
class AuthorUpdate(BaseModel):
name: str
date_of_birth: date
def update_author(
cursor: pg.Cursor,
update: AuthorUpdate,
id_: int,
) -> None:
pg.update_many(
cursor=cursor,
update=e.author,
set_=((e.name, e.date_of_birth), update),
where=e.id == id_,
)
from datetime import date
from pydantic import BaseModel
import pgcrud as pg
from pgcrud import e
class AuthorUpdate(BaseModel):
name: str
date_of_birth: date
async def update_author(
cursor: pg.a.Cursor,
update: AuthorUpdate,
id_: int,
) -> None:
await pg.a.update_many(
cursor=cursor,
update=e.author,
set_=((e.name, e.date_of_birth), update),
where=e.id == id_,
)
Delete
The delete_many method is used to delete records from a table.2 You need to specify from which table you want to delete the records.
The where condition specifies which records should be deleted. To delete a single record, you must ensure that the where condition
targets a single record.
License
pgcrud is released under the MIT License.