Sqlalchemy module for nestipy framework.
Project description
Description
Nestipy is a Python framework built on top of FastAPI that follows the modular architecture of NestJS
Under the hood, Nestipy makes use of FastAPI, but also provides compatibility with a wide range of other libraries, like Blacksheep, allowing for easy use of the myriad of third-party plugins which are available.
Getting started
pip install nestipy-cli
nestipy new my_app
cd my_app
nestipy start --dev
├── src
│ ├── __init__.py
├── app_module.py
├── app_controller.py
├── app_service.py
├── main.py
├── requirements.txt
├── README.md
Documentation
View full documentation from here.
SQLAlchemy to Pydantic
import asyncio
import uuid
from typing import List
from sqlalchemy import String, ForeignKey, select, create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import declarative_base, Mapped, mapped_column, relationship, DeclarativeBase, Session
from nestipy_alchemy import SqlAlchemyPydanticLoader
from nestipy_alchemy import SqlAlchemyPydanticMapper
Base: DeclarativeBase = declarative_base()
class Employee(Base):
__pydantic_name__ = "EmployeeSchema"
__tablename__ = "employee"
__pydantic_exclude__ = ["password_hash"]
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(255), nullable=False)
password_hash: Mapped[str] = mapped_column(String(255), nullable=False)
department_id: Mapped[str] = mapped_column(String(36), ForeignKey("department.id"))
department: Mapped["Department"] = relationship("Department", back_populates="employees")
class Department(Base):
__pydantic_name__ = "DepartmentSchema"
__tablename__ = "department"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(255), nullable=False)
employees: Mapped[List[Employee]] = relationship("Employee", back_populates="department")
sqlalchemy_pydantic_mapper = SqlAlchemyPydanticMapper()
async def test_with_async():
sqlalchemy_pydantic_mapper.type(model=Employee)(type("Employee", (), {}))
engine = create_async_engine("mysql+asyncmy://root@localhost:3306/alchemy", echo=True)
async with engine.begin() as conn:
pass
async with AsyncSession(bind=engine) as session:
sqlalchemy_pydantic_loader = SqlAlchemyPydanticLoader(
_mapper=sqlalchemy_pydantic_mapper,
async_session_callback=lambda: session
)
stmt = select(Employee)
result = await session.execute(stmt)
all_employee = result.scalars().all()
for employee in all_employee:
json = await sqlalchemy_pydantic_loader.load(employee, depth=5)
print(json.model_dump(mode="json"))
await session.commit()
def test_with_sync():
engine = create_engine("mysql+pymysql://root@localhost:3306/alchemy", echo=True)
# Base.metadata.drop_all(bind=engine)
# Base.metadata.create_all(bind=engine)
sqlalchemy_pydantic_mapper.type(model=Employee)(type("Employee", (), {}))
with Session(bind=engine, expire_on_commit=False) as session:
# dep = Department(name="Test", employees=[Employee(name="Employee1", password_hash="test")])
# session.add(dep)
sqlalchemy_pydantic_loader = SqlAlchemyPydanticLoader(
_mapper=sqlalchemy_pydantic_mapper,
session=session
)
stmt = select(Employee)
result = session.execute(stmt)
all_employee = result.scalars().all()
for employee in all_employee:
json = sqlalchemy_pydantic_loader.load_sync(employee, depth=5)
print(json.model_dump(mode="json"))
session.commit()
# print(sqlalchemy_pydantic_mapper.mapped_types.get("DepartmentSchema"))
if __name__ == "__main__":
asyncio.run(test_with_async())
# test_with_sync()
Support
Nestipy is an MIT-licensed open source project. It can grow thanks to the sponsors and support from the amazing backers. If you'd like to join them, please [read more here].
Stay in touch
- Author - Tsiresy Mila
License
Nestipy is MIT licensed.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
nestipy_alchemy-0.1.12.tar.gz
(7.7 kB
view hashes)
Built Distribution
Close
Hashes for nestipy_alchemy-0.1.12-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 1fbdae3594fb96d7c8332745a51fcb171a225d8dfb404f30a04e95d0262ff57b |
|
MD5 | 6e95d4e7420d2c9e2e64e092ab64f874 |
|
BLAKE2b-256 | 55ad24415f541b42b683cb2b76432006a846026c46fd8f3f271d76feb8ea1292 |