安装fastapi

安装框架

pip install fastapi

安装uvicorn作为服务器启动项目

pip install  "uvicorn[standard]" 

用fastapi写一个简单的路由

fastapi定义路由有两种方式

a:在fastapi实例上定义路由

from fastapi import FastAPIapp = FastAPI()@app.get('/')
def root():return {"message": "Hello word"}

这种方式适用于小型应用，代码简洁,所有路由定义都在同一个文件中但是路由定义不能轻易复用，除非复制粘贴代码,每个路由都需要单独定义路径

b:使用 APIRouter 定义路由

from fastapi import APIRouter, FastAPI# 创建一个 APIRouter 实例
router = APIRouter()# 在 APIRouter 上定义路由
@router.get('/')
def root():return {"message": "Hello World"}# 创建 FastAPI 应用实例
app = FastAPI()# 将 APIRouter 添加到 FastAPI 应用中
app.include_router(router)

这种方式更适合大型应用，因为它允许你将路由逻辑组织到不同的模块中，从而提高代码的可维护性和复用性,路由定义可以分散在多个文件中，每个文件可以定义一个或多个 APIRouter，然后在主应用中通过 include_router 方法将这些路由器合并到一起。这种方式更适用于大型应用，有助于提高代码的组织性和可维护性

启动fastapi项目

命令:

        uvicorn 启动文件:app --reload

fastapi自带接口文档可以通过下面的路由查看

        FastAPI - Swagger UIhttp://127.0.0.1:8000/docs

路由传参

from fastapi import FastAPI, APIRouter# 创建一个 APIRouter 实例
router = APIRouter()# 在 APIRouter 上定义路由
@router.get('/')
def root():return {"message": "Hello World"}@router.get('/items/{item_id}')
def read_item(item_id: int):return {"item_id": item_id}# 创建 FastAPI 应用实例
app = FastAPI()# 将 APIRouter 添加到 FastAPI 应用中
app.include_router(router)INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [16796] using WatchFiles
INFO:     Started server process [3836]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     127.0.0.1:51257 - "GET / HTTP/1.1" 200 OK
INFO:     127.0.0.1:51258 - "GET / HTTP/1.1" 200 OK
INFO:     127.0.0.1:51288 - "GET /items/1 HTTP/1.1" 200 OK

这样我们可以实现简单的路由传参，请求就是最后一个请求,如果类型不匹配则会报错

INFO:     127.0.0.1:51467 - "GET /items/aaa HTTP/1.1" 422 Unprocessable Entity
这个就代表我们传的参数跟实际要求类型不匹配

 在上面的例子中,item_id 被声明为int类型,FastApi通过上面的类型声明提供了对请求的自动解析,同时还提供了数据校验功能,在路由上的参数通过http都会传为字符串,fastapi会将这个类型转为我们指定的类型,如果无法转换则会报错

这个时候要注意一个问题,当我们要写一个查询所有跟查询单个接口的时候,要按照下面的写

@router.get('/items/{item_id}', summary='查询单个')
def read_item(item_id: int):return {"item_id": item_id}@router.get('/items', summary='查询所有名称')
def read_items():return {"item_id": "all"}

这样他会去匹配不同的路由，但是如果按照下面的写法就会报错

@router.get('/items/all', summary='查询所有名称')
def read_items():return {"item_id": "all"}

这个时候就是我们上面说的fastapi会自动帮我们解析校验,按照下面的写法,我们的路由会是items/all 那么会先触发上面查询单个的路由，然后发现all字符串无法转int直接报错

{"detail": [{"type": "int_parsing","loc": ["path","item_id"],"msg": "Input should be a valid integer, unable to parse string as an integer","input": "all"}]
}

所以一定要注意路由的写法 

URL传参

url请求参数是通过url地址携带

http://127.0.0.1:8000/item/?s=1&d=2

?后面就属于url传参

@router.get('/item', summary='搜索功能')
def get_item(id: int = Query(None, title='id', description='id'),name: str = Query(None, title='name', description='name')):"""Query 表示专门用来对FastAPI的url传参进行限制,default 表示这个参数可以为None"""return {"id": id, "name": name}INFO:     127.0.0.1:52601 - "GET /item HTTP/1.1" 200 OK
INFO:     127.0.0.1:52603 - "GET /item?id=1 HTTP/1.1" 200 OK
INFO:     127.0.0.1:52795 - "GET /item?id=1&name=zh HTTP/1.1" 200 OK

 

 我们还可以传其他的类型,如列表如下展示:

@router.delete('/item', summary='删除多个')
def delete_item(id_list: List[int] = Query([], title='id_list', description='id列表')):"""这里我们传过来的参数就是一个列表了"""return {"id_list": id_list}

请求如下图所示

INFO:     127.0.0.1:52508 - "DELETE /item?id_list=1&id_list=2&id_list=3 HTTP/1.1" 200 OK

我们这里限制了,列表里面必须是int类型,如果有一个不为int看会是什么样的

Please correct the following validation errors and try again.

• For 'id_list' at path '[1]': Value must be an integer. 在文档里面会直接报错的,还会给出索引位置不符合要求

SQLAIchemy 2.0

SQLAIchemy 是python生态系统中最流行的ORM，他的设计分为两部分--底层的Core和上层的传统ORM，在python乃至其他语言的大多数ORM中,都没有实现很好的分层涉及,比如django的ORM,数据库链接和ORM本身完全混在一起,他可以在python的框架都可以使用,django的ORM只能在django中使用

 

  SQLALchemy是python中一个通过ORM操作数据库的框架

SQLALchemy对象关系映射器提供了一种方法,用于将用户定义的python类与数据库表相关联,并将这些类(对象)的示例与其对应表中的行相关联,它包括一个透明地同步对象及其相关行之间状态地所有变化地系统,称为 工作单元 ，根据用户定义地类及其定义地彼此之间地关系表达数据库查询的系统

可以让我们使用类和对象的方式操作数据库,从而从繁琐的sql语句中解脱出来

安装sqlalchemy

pip install sqlalchemy

数据库引擎

任何SQLALchemy应用程序的开始都是一个名为Engine,此对象充当连接到特定数据库的中心源,提供工厂和成为connection pool 对于这些数据库连接,引擎通常是一个只为特定数据库服务器创建一次的全局对象,并使用一个一个URL字符串进行配置,该字符串将描述如何连接到数据库主机或后端

总体结构如下:

sqlclchemy使用create_engine()函数从URL生成一个数据库引擎对象,比如:

import sqlalchemyengine = sqlalchemy.create_engine('url')

我们看一下源码，就能看到他这里也是可以看出来的

def create_engine(url: Union[str, _url.URL], **kwargs: Any) -> Engine:"""Create a new :class:`_engine.Engine` instance.

        pass

1.支持的数据库

URL通常可以包括用户名,密码, 主机名，数据库名以及用于其他配置的可选关键字参数，主题格式如下：

dialect+driver://username:password@host:port/database

dialect:数据库类型

driver:数据库驱动

username:数据库用户名

password:数据库密码

host:数据库主机地址

port:数据库端口

database:要连接的数据库名称

如,我们想要连接mysql的数据库就要按照下面写:

但是使用pymysql就需要先安装这个数据库驱动

pip install  pymysql 或者也可以使用pip install mysqlclient 这个时候就要去改对应的数据库驱动

database_url = 'mysql+pymysql://root:password@localhost:3306/test_db?charset=utf8'

 2.数据库引擎参数

def create_engine(url: Union[str, _url.URL], **kwargs: Any) -> Engine:

他其中有几个参数比较重要:

echo=False: 如果为True,引擎将记录所有语句以及repr()其参数列表的默认日志处理程序

future=True: 使用2.0 样式Engine和Connection API

logging_name:将在sqlclchemy.engine 记录器中生成的日志记录的 name 字段中使用的字符串标识符

pool_size=5:连接池的大小,默认为5个,设置为0时表示无连接限制

pool_recycle=3600:设置实际限制数据库自动断开连接

pool_timeout=30:连接超时时间,默认为30秒,超过实际的连接都会连接失败

示例:

import sqlalchemy# 数据库 URL 格式
# dialect+driver://username:password@host:port/database# 示例数据库 URL
database_url = 'mysql+pymysql://username:password@localhost:3306/database_name'# 创建引擎
engine = sqlalchemy.create_engine(database_url)# 测试连接
try:with engine.connect() as connection:print("数据库连接成功")
except Exception as e:print(f"连接失败: {e}")