pyNest,把FastAPI程序寫出Spring的風格
- 快速入門
- 1、安裝pyNest
- 2、創建項目
- 3、編寫app_module.py
- 4、編寫app_service.py
- 5、編寫app_controller.py
- 6、編寫main.py
- 7、啟動程序
- 核心概念
- 1、Modules
- 2、Controllers
- 3、Providers
- 4、ORM Provider
NestJS是風靡于Node.js圈的web框架,它深度借鑒了Spring Boot的設計思路,但又比Spring Boot小巧靈活。
對于轉戰于python圈的小伙伴來說,如果可以把Spring Boot搬運到python,那將是一件非常有價值的事情,于是就有了pyNest這個“小酷庫”——它站在FastAPI的基礎,在python中實現了NestJS風格,讓你的代碼有了Spring的味道。本文我們就一起來試試pyNest這個“小酷庫”。
快速入門
1、安裝pyNest
pip install pynest-api
2、創建項目
my_pynest_project/
├── src/
│ ├── __init__.py
│ ├── app_module.py
│ ├── app_controller.py
│ ├── app_service.py
├── main.py
├── requirements.txt
└── README.md
3、編寫app_module.py
from nest.core import Module, PyNestFactory
from .app_controller import AppController
from .app_service import AppService@Module(controllers=[AppController],providers=[AppService],
)
class AppModule:passapp = PyNestFactory.create(AppModule,description="This is my PyNest app",title="My App",version="1.0.0",debug=True,
)http_server = app.get_server()
4、編寫app_service.py
from nest.core import Injectable@Injectable
class AppService:def __init__(self):self.app_name = "MyApp"self.app_version = "1.0.0"def get_app_info(self):return {"app_name": self.app_name, "app_version": self.app_version}
5、編寫app_controller.py
from nest.core import Controller, Get
from .app_service import AppService@Controller("/")
class AppController:def __init__(self, service: AppService):self.service = service@Get("/")def get_app_info(self):return self.service.get_app_info()
6、編寫main.py
import uvicorn
from src.app_module import http_serverif __name__ == "__main__":uvicorn.run(http_server, host="0.0.0.0", port=8000, reload=True)
7、啟動程序
python main.py
核心概念
PyNest 主要包括Modules、Controllers、Providers三個核心概念,其中:
- Module負責組織和管理 Controller 和 Provider;
- Controller 負責接收請求,調用注入的 Provider 來完成業務邏輯;
- Provider 是服務/工具,也可以互相注入組合;
- 模塊之間通過 imports/exports 連接起來,實現模塊解耦和共享。
它們的關系如下圖:
1、Modules
模塊(Modules)是一個帶有@Module()
裝飾器注解的類。@Module()
裝飾器提供 PyNest 用于組織應用程序結構的元數據。
在pyNest中,每個應用程序至少有一個模塊,即根模塊。根模塊是 PyNest 構建應用程序圖的起點,而應用程序圖是 PyNest 用于解析模塊和提供程序關系及依賴關系的內部數據結構。雖然非常小的應用程序理論上可能只有一個根模塊,但這并不是典型情況。強烈建議使用模塊作為組織組件的有效方法。
模塊定義了以下元數據:
- controllers:控制器列表
- providers:服務提供者(通常是業務邏輯)
- imports:導入其他模塊
- exports:暴露自己的部分 provider 給其他模塊使用
2、Controllers
控制器(Controllers )是一個帶有@Controller()
裝飾器注解的類。該類負責處理傳入的請求并向客戶端返回響應。控制器在應用程序中注冊路由并管理請求和響應對象,有效地充當客戶端與應用程序交互的網關。
pyNest 支持 5 種 http 方法——Get、Post、Put、Delete、Patch。由于 Pynest 是 FastAPI 的抽象,因此我們可以像在 FastAPI 中一樣使用這些方法。
以下為一個完整的 CRUD 示例:
from nest.core import Controller, Get, Post, Put, Delete
from .book_service import BookService
from .book_models import Book@Controller('/books')
class BooksController:def __init__(self, book_service: BookService):self.book_service = book_service@Get('/')def get_books(self):return self.book_service.get_books()@Get('/:book_id')def get_book(self, book_id: int):return self.book_service.get_book(book_id)@Post('/')def add_book(self, book: Book):return self.book_service.add_book(book)@Put('/:book_id')def update_book(self, book_id: int , book: Book):return self.book_service.update_book(book_id, book)@Delete('/:book_id')def delete_book(self, book_id: int):return self.book_service.delete_book(book_id)
3、Providers
服務(Providers),是一個帶有@Injectable
裝飾器注解的類,是 PyNest 應用程序的基礎。它們處理業務邏輯和其他功能,充當可注入到其他組件(例如控制器)的依賴項。本指南通過各種示例講解了什么是提供程序、如何使用它們、如何在模塊中導入和導出它們,以及如何將它們注入到控制器和其他服務中。
定義Provider
from nest.core import Injectable@Injectable
class LoggerService:def log(self, message: str):print(f"LOG: {message}")
在例子中,LoggerService
使用@Injectable
裝飾器將類定義為提供程序。現在可以將此類注入到同一模塊內的其他類中,或將其導出以供其他模塊使用。
在模塊中注冊Provider
from nest.core import Module
from .s3_service import S3Service
from src.providers.logger.logger_service import LoggerService
from src.providers.logger.logger_module import LoggerModule@Module(providers=[S3Service, LoggerService],exports=[S3Service],imports=[LoggerModule],
)
class S3Module:pass
在例子中:S3Module 定義了一個提供 S3Service 的模塊,并將其導出以供其他模塊使用。它還導入了 LoggerModule 以使用 LoggerService。
將Provider注入Controller
from nest.core import Controller, Post
from .s3_service import S3Service@Controller('/s3')
class S3Controller:def __init__(self, s3_service: S3Service):self.s3_service = s3_service@Post('/upload')def upload_file(self, file_name: str):self.s3_service.upload_file(file_name)return {"message": "File uploaded successfully!"}
在這個例子中,S3Service通過構造函數注入到了S3Controller中。
將Provider注入其他Provider
from nest.core import Injectable
from src.providers.logger.logger_service import LoggerService@Injectable
class S3Service:def __init__(self, logger_service: LoggerService):self.logger_service = logger_servicedef upload_file(self, file_name: str):print(f"Uploading {file_name}")self.logger_service.log(f"Uploaded file: {file_name}")
4、ORM Provider
pyNest內置了基于SQLAlchemy的數據庫連接組件,它包括了OrmProvider( 同步)、AsyncOrmProvider( 異步)兩個服務。
以下是AsyncOrmProvider的使用示例:
config.py
from nest.core.database.orm_provider import AsyncOrmProvider
import os
from dotenv import load_dotenvload_dotenv()config = AsyncOrmProvider(db_type="postgresql",config_params=dict(host=os.getenv("POSTGRESQL_HOST", "localhost"),db_name=os.getenv("POSTGRESQL_DB_NAME", "default_nest_db"),user=os.getenv("POSTGRESQL_USER", "postgres"),password=os.getenv("POSTGRESQL_PASSWORD", "postgres"),port=int(os.getenv("POSTGRESQL_PORT", 5432)),)
)
app_schema.py
from src.config import config
from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_columnclass Example(config.Base):__tablename__ = "example"id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)name: Mapped[str] = mapped_column(String, unique=True)
app_service.py
from .examples_model import Examples
from .examples_entity import Examples as ExamplesEntity
from src.config import config
from nest.core.decorators.database import async_db_request_handler
from nest.core import Injectable
from sqlalchemy import select@Injectable
class ExamplesService:def __init__(self):self.orm_config = configself.session = self.orm_config.get_session@async_db_request_handlerasync def add_examples(self, examples: Examples):examples_entity = ExamplesEntity(**examples.dict())async with self.session() as session:session.add(examples_entity)await session.commit()return examples_entity.id@async_db_request_handlerasync def get_examples(self):query = select(ExamplesEntity)async with self.session() as session:result = await session.execute(query)return result.scalars().all()
app_controller.py
from nest.core import Controller, Get, Postfrom .examples_service import ExamplesService
from .examples_model import Examples@Controller("examples")
class ExamplesController:def __init__(self, service: ExamplesService):self.service = service@Get("/")async def get_examples(self):return await self.service.get_examples()@Post("/")async def add_examples(self, examples: Examples):return await self.service.add_examples(examples)
app_module.py
from .config import config
from .example.example_module import ExampleModulefrom .app_controller import AppController
from .app_service import AppServicefrom nest.core import Module, PyNestFactory@Module(imports=[ExampleModule],controllers=[AppController],providers=[AppService],
)
class AppModule:passapp = PyNestFactory.create(AppModule, description="This is my FastAPI app drive by Async ORM Engine", title="My App",version="1.0.0", debug=True)http_server = app.get_server()@http_server.on_event("startup")
async def startup():await config.create_all()