测试平台接口开发日记_from pydantic import anyhttpurl 作用

上一节的基础的工程目录已经完成，本次是开始代码的入手，首先是先从数据库的基类封装和配置文件入手。

配置文件处理

---

在 src 目录下新建 dev.env 配置文件，主要放置一些配置信息，后续的 redis、服务器启动配置等，具体内容如下：

# Mysql数据库连接信息
MYSQL_SERVER=10.10.12.199
MYSQL_PORT=3307
MYSQL_USER=root
MYSQL_PASSWORD=123456
MYSQL_DB=back_end

因为想要把配置文件的数据库信息拼接起来。如下显示：

mysql+pymysql://root:123456@10.10.12.199:3307/back_end?charset=utf8

所以新建一个 cabinet 目录，并在 code_pydantic.py 封装了一个DatabasesDsn类，它的主要功能就是验证数据库地址。DatabasesDsn类是继承pydantic中AnyUrl类来实现的，具体截图和代码如下：

from pydantic import AnyUrl
 
 
__all__ = [
    "DatabasesDsn"
]
 
 
class DatabasesDsn(AnyUrl):
    allowed_schemes = {
        'mysql+pymysql',
        'postgresql',
        'postgresql+asyncpg',
        'postgresql+pg8000',
        'postgresql+psycopg',
        'postgresql+psycopg2',
        'postgresql+psycopg2cffi',
        'postgresql+py-postgresql',
        'postgresql+pygresql',
    }

注释信息：allowed_schemes属性的作用就是重写了AnyUrl类中的allowed_schemes属性，其父类内部的方法未作任何重写操作

在 conf 目录中，新建 confing.py 文件并封装一个 Settings 类，主要的功能就是读取配置文件中的数据，最后新建settings对象，具体代码如下：

import os
from typing import (
    Optional,
    Dict,
    Any
)
from pydantic import (
    BaseSettings,
    validator
)
from src.app.public.dantic import DatabasesDsn
 
__all__ = [
    "settings"
]
 
 
class Settings(BaseSettings):
 
    # 数据库连接信息
    MYSQL_SERVER: str
    MYSQL_PORT: str
    MYSQL_USER: str
    MYSQL_PASSWORD: str
    MYSQL_DB: str
    SQLALCHEMY_DATABASE_URI: Optional[DatabasesDsn] = None
 
    # validator是指一个用于验证数据是否符合规则的函数。# 它可以用于验证输入的数据是否为字符串、整数、浮点数等等，并在不符合规则时抛出相应的异常。@validator("SQLALCHEMY_DATABASE_URI")
    def assemble_db_connection(cls, v: Optional[str], values: Dict[str, Any]):
        if isinstance(v, str):
            return v
        return DatabasesDsn.build(
            scheme="mysql+pymysql",
            user=values.get("MYSQL_USER"),
            password=values.get("MYSQL_PASSWORD"),
            host=values.get("MYSQL_SERVER"),
            port=values.get("MYSQL_PORT"),
            path=f'/{values.get("MYSQL_DB")}'
        )
 
 
class SettingsDevd(Settings):
    # 开发环境读取配置
    class Config:
        case_sensitive = True
        env_file = "../dev.env"
        env_file_encoding = "utf-8"
 
 
settings = SettingsDevd()

数据库操作类封装

---

这里把public目录(package包)名录名称修改为core目录(package包)，然后在 core 中新建一个 db 目录(package包)

在 db 目录中新建 base_calss.py 文件并封装一个 Base 类，主要功能是用于数据库模型需要继承的基类。具体代码如下：

from typing import Any
from sqlalchemy.ext.declarative import as_declarative, declared_attr
 
 
# as_declarative将一个类或模型转换为声明式模式。# 这个函数的作用是将类或模型中的所有属性转换为类属性或实例属性，同时将所有的方法转换为非懒加载方法。# 转换完成后，该类或模型就可以像声明式的对象一样使用了。@as_declarative()
class Base:
    __name__: str
 
    # 函数可以用来获取模型中声明的属性。它返回一个字典，其中键是属性名称，值是属性对应的类属性或实例属性。# 这个函数通常用在ORM模型中，可以用来获取模型中被声明的属性，并将其添加到调用函数的参数中。@declared_attr
    def __tablename__(cls):
        return cls.__name__.lower()

在 db 目录中新建 session_context.py 文件并封装一个 SessionContextManager 类，主要功能是自定义一个上下文管理器。具体代码如下：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from src.conf.config import settings
 
 
class SessionContextManager:
    def __init__(self) -> None:
        # 创建一个持久对象
        db_session = sessionmaker(
            autocommit=False, autoflush=False,
            # 连接数据库
            bind=create_engine(
                url=settings.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True
            )
        )
        # 新建一个session对象
        self._db = db_session()
 
    def __enter__(self):
        return self._db
 
    def __exit__(self, exc_type, exc_value, traceback):
        return self._db.close()

注释信息：在 session_context.py 中我们引入了 settings 对象，settings 的主要功能是读取配置文件中数据，而 settings.SQLALCHEMY_DATABASE_URI 是获取数据库的连接地址。

在 db 目录中新建 session.py 文件并封装一个 session 异步函数和 session_commit 往数据提交数据的函数，主要的功能是用于操作数据库连接初始化，调用方法操作数据库数据，具体代码如下：

from sqlalchemy.orm import Session
from src.app.core.db.session_context import SessionContextManager
 
__all__ = [
    "session",
    "session_commit",
]
 
 
async def session():
    with SessionContextManager() as db:
        yield db
 
 
def session_commit(db: Session, *, datas):
    # 往数据库添加数据
    db.add(datas)
    # 往数据库提交数据
    db.commit()
    # 刷新提交的数据
    db.refresh(datas)
    return datas

在 db 目录中新建 base.py 文件,主要的功能是用于集合当前目录下的引包，具体代码如下：

from src.app.public.db.base_class import Base
from src.app.public.db.session import session

最后： 为了回馈铁杆粉丝们，我给大家整理了完整的软件测试视频学习教程，朋友们如果需要可以自行免费领取【保证100%免费】

软件测试面试文档

我们学习必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有字节大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。