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()