阿里巴巴国际站运营_深圳品牌家政公司排行榜_重庆做网络优化公司电话_兰州网站seo

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