我尽量用代码的形式给大家展示

# 1,创建异步引擎，
# 2，定义模型类: 基类 + 表对应的模型类
# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
from datetime import datetime

from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

# 1,创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=10  # 允许额外的连接数
)


# 2，定义模型类: 基类 + 表对应的模型类
# 基类: 创建时间、更新时间； 书籍表: id,书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间")


class Book(Base):
    __tablename__ = "book"  # 建一个名为book的表
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    book_name: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[int] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_table():
    # 获取异步引擎，创建事务 - 建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建


@app.on_event("startup")
async def startup():
    await create_table()


@app.get("/")
async def root():
    return {"message": "Hello World"}


# 需求：查询功能的接口，查询图书 -> 依赖注入: 创建依赖项获取数据库会话 + Depends 注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚e
            raise
        finally:
            await session.close()


@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
    # 查询
    result = await db.execute(select(Book))
    book = result.scalars().all()  # scalars().all()获取所有的数据
    return book

"""
数据库操作 - 查询
核心语句: await db.execute( select(模型类)), 返回一个ORM对象

获取所有数据
    scalars().all()
    
获取单条数据
    scalars().first() -> 获取第一条
    get(模型类, 主键值)
"""
from datetime import datetime

from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

# 1,创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=10  # 允许额外的连接数
)


# 2，定义模型类: 基类 + 表对应的模型类
# 基类: 创建时间、更新时间； 书籍表: id,书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间")


class Book(Base):
    __tablename__ = "book"  # 建一个名为book的表
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    book_name: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[int] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_table():
    # 获取异步引擎，创建事务 - 建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建


@app.on_event("startup")
async def startup():
    await create_table()


@app.get("/")
async def root():
    return {"message": "Hello World"}


# 需求：查询功能的接口，查询图书 -> 依赖注入: 创建依赖项获取数据库会话 + Depends 注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚e
            raise
        finally:
            await session.close()


@app.get("/book/book_list")
async def get_book(db: AsyncSession = Depends(get_database)):
    # result = await db.execute(select(Book))
    # book = result.scalars().all()  # 获取所有的数据
    # book = result.scalars().first()  # 获取第一条数据
    book = await db.get(Book,5)
    return book


"""
数据库操作 - 查询
核心语句: await db.execute( select(模型类)), 返回一个ORM对象

获取所有数据
    scalars().all()
    
获取单条数据
    scalars().first() -> 获取第一条
    get(模型类, 主键值)
"""
"""
数据库操作 - 查询条件
select(Book).where(条件, 条件2, ...)
条件:
    比较判断: ==; >; <; >=; <= 等
    模糊查询: like()
    与非查询: &; |; ~
    包含查询: in_()
"""
from datetime import datetime

from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

# 1,创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=10  # 允许额外的连接数
)


# 2，定义模型类: 基类 + 表对应的模型类
# 基类: 创建时间、更新时间； 书籍表: id,书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间")


class Book(Base):
    __tablename__ = "book"  # 建一个名为book的表
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    book_name: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[int] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_table():
    # 获取异步引擎，创建事务 - 建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建


@app.on_event("startup")
async def startup():
    await create_table()


@app.get("/")
async def root():
    return {"message": "Hello World"}


# 需求：查询功能的接口，查询图书 -> 依赖注入: 创建依赖项获取数据库会话 + Depends 注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚e
            raise
        finally:
            await session.close()


@app.get("/book/book_list/{book_id}")
async def get_search_book(book_id: int, db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(book_id == Book.id))  # 执行数据库查询：根据book_id查找匹配的图书记录
    book = result.scalar_one_or_none()
    return book


# 需求: 查询价格大于20的书籍
@app.get("/book/search_book")
async def search_book(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book).where(Book.price >= 20))
    books = result.scalars().all()
    return books

"""
数据库操作 - 查询条件
select(Book).where(条件, 条件2, ...)
条件:
    比较判断: ==; >; <; >=; <= 等
    模糊查询: like()
    与非查询: &; |; ~
    包含查询: in_()
"""

"""
查询条件 - 模糊查询
模糊查询：like()
% : 零个、一个或多个字符
_ : 一个单个字符
"""
from datetime import datetime

from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

# 1,创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=10  # 允许额外的连接数
)


# 2，定义模型类: 基类 + 表对应的模型类
# 基类: 创建时间、更新时间； 书籍表: id,书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间")


class Book(Base):
    __tablename__ = "book"  # 建一个名为book的表
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    book_name: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[int] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_table():
    # 获取异步引擎，创建事务 - 建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建


@app.on_event("startup")
async def startup():
    await create_table()


@app.get("/")
async def root():
    return {"message": "Hello World"}


# 需求：查询功能的接口，查询图书 -> 依赖注入: 创建依赖项获取数据库会话 + Depends 注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚e
            raise
        finally:
            await session.close()


# 需求: 出版社以 起 开头 %、_
@app.get("/book/book_list")
async def get_book(db: AsyncSession = Depends(get_database)):
    """
    like() 模糊查询:
            % : 任意个字符;
             _ : 一个单个字符
    """
    # result = await db.execute(select(Book).where(Book.publisher.like("起_")))  -> []
    # & | ~ 与非
    # result = await db.execute(select(Book).where((Book.publisher.like("起%")) | (Book.price > 12)))
    # 需求: 书籍id 列表，数据库里面的 id 如果在 书籍id列表里面 就返回
    id_list = [2,4,6,8,11]
    result = await db.execute(select(Book).where(Book.id.in_(id_list)))
    book = result.scalars().all()
    return book

"""
查询条件 - 模糊查询
模糊查询：like()
% : 零个、一个或多个字符
_ : 一个单个字符
"""

"""
数据库操作 - 聚合查询
聚合计算： func.方法(模型类.属性)
count：统计行数量
avg：求平均值
max: 求最大值
min: 求最小值
sum: 求和
"""
from datetime import datetime

from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

# 1,创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=10  # 允许额外的连接数
)


# 2，定义模型类: 基类 + 表对应的模型类
# 基类: 创建时间、更新时间； 书籍表: id,书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间")


class Book(Base):
    __tablename__ = "book"  # 建一个名为book的表
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    book_name: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[int] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_table():
    # 获取异步引擎，创建事务 - 建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建


@app.on_event("startup")
async def startup():
    await create_table()


@app.get("/")
async def root():
    return {"message": "Hello World"}


# 需求：查询功能的接口，查询图书 -> 依赖注入: 创建依赖项获取数据库会话 + Depends 注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚e
            raise
        finally:
            await session.close()


@app.get('/book/juhe')
async def get_book(db: AsyncSession = Depends(get_database)):
    # result = await db.execute(select(func.count(Book.id)))  # 统计数据库中数据的条数
    # result = await db.execute(select(func.sum(Book.price)))  # 计算价格的总和
    # result = await db.execute(select(func.avg(Book.price)))  # 计算价格的平均值
    result = await db.execute(select(func.max(Book.price)))  # 得到最高的书籍价格
    num = result.scalar()  # 用来提取一个数值  -> 标量值
    return num

"""
数据库操作 - 聚合查询
聚合计算： func.方法(模型类.属性)
count：统计行数量
avg：求平均值
max: 求最大值
min: 求最小值
sum: 求和
"""

"""
数据库查询操作 - 分页查询
分页查询： select().offset().limit()
    offset：跳过的记录数
    limit: 返回的记录数
    offset值 = (当前页码 - 1) * 每页数量
"""
from datetime import datetime

from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

# 1,创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=10  # 允许额外的连接数
)


# 2，定义模型类: 基类 + 表对应的模型类
# 基类: 创建时间、更新时间； 书籍表: id,书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间")


class Book(Base):
    __tablename__ = "book"  # 建一个名为book的表
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    book_name: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[int] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_table():
    # 获取异步引擎，创建事务 - 建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建


@app.on_event("startup")
async def startup():
    await create_table()


@app.get("/")
async def root():
    return {"message": "Hello World"}


# 需求：查询功能的接口，查询图书 -> 依赖注入: 创建依赖项获取数据库会话 + Depends 注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚e
            raise
        finally:
            await session.close()


@app.get("/book/book_list")
async def get_book(
        page: int,
        page_size: int = 2,
        db: AsyncSession = Depends(get_database)
):
    # offset: 跳过的数量  limit: 每页的数据量
    skip = (page - 1) * page_size
    result = await db.execute(select(Book).offset(skip).limit(page_size))
    book = result.scalars().all()
    return book

"""
数据库查询操作 - 分页查询
分页查询： select().offset().limit()
    offset：跳过的记录数
    limit: 返回的记录数
    offset值 = (当前页码 - 1) * 每页数量
"""


"""
数据库操作 - 新增

核心步骤： 定义ORM对象 -> 添加对象到事务： add(对象) -> commit提交到数据库
"""
# ** 字典键值对取值，__dict__变成字典格式
from datetime import datetime

from fastapi import FastAPI, Depends
from sqlalchemy import DateTime, func, String, Float, select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from pydantic import BaseModel

app = FastAPI()

# 1,创建异步引擎
ASYNC_DATABASE_URL = "mysql+aiomysql://root:123456@localhost:3306/fastapi_first?charset=utf8"
async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,  # 可选，输出SQL日志
    pool_size=10,  # 设置连接池活跃的连接数
    max_overflow=10  # 允许额外的连接数
)


# 2，定义模型类: 基类 + 表对应的模型类
# 基类: 创建时间、更新时间； 书籍表: id,书名，作者，价格，出版社
class Base(DeclarativeBase):
    create_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), comment="创建时间")
    update_time: Mapped[datetime] = mapped_column(DateTime, insert_default=func.now(), default=func.now(), onupdate=func.now(), comment="更新时间")


class Book(Base):
    __tablename__ = "book"  # 建一个名为book的表
    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    book_name: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[int] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


# 3，建表: 定义函数建表 -> FastAPI 启动的时候调用建表的函数
async def create_table():
    # 获取异步引擎，创建事务 - 建表
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)  # Base 模型类的元数据创建


@app.on_event("startup")
async def startup():
    await create_table()


@app.get("/")
async def root():
    return {"message": "Hello World"}


# 需求：查询功能的接口，查询图书 -> 依赖注入: 创建依赖项获取数据库会话 + Depends 注入路由处理函数
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,  # 绑定数据库引擎
    class_=AsyncSession,  # 指定会话类
    expire_on_commit=False  # 提交后会话不过期，不会重新查询数据库
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session  # 返回数据库会话给路由处理函数
            await session.commit()  # 提交事务
        except Exception:
            await session.rollback()  # 有异常，回滚e
            raise
        finally:
            await session.close()


# 需求： 用户输入图书信息(id,书名，作者，价格，出版社) -> 新增
# 用户输入 -> 参数 -> 请求体
class BookBase(BaseModel):
    id: int
    book_name: str
    author: str
    price: int
    publisher: str


@app.post("/book/book_list")
async def add_book(book: BookBase, db: AsyncSession = Depends(get_database)):
    # ORM对象 -> add -> commit
    book_obj = Book(**book.__dict__)  # ** 字典键值对取值，__dict__变成字典格式
    db.add(book_obj)
    await db.commit()
    return book

"""
数据库操作 - 新增

核心步骤： 定义ORM对象 -> 添加对象到事务： add(对象) -> commit提交到数据库
"""

# fastapi # 学习 # python
Logo
腾讯云开发者社区

腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。

更多推荐

Elasticsearch复杂数据类型终极指南：从入门到精通

Elasticsearch作为功能强大的搜索引擎，支持多种复杂数据类型，让开发者能够灵活处理各种结构化和非结构化数据。本文将带你全面了解Elasticsearch中的复杂数据类型，从基础概念到实际应用，助你轻松掌握数据建模的核心技巧。## 内部对象：构建层级化数据结构在Elasticsearch中，对象类型（Object）是最基础的复杂数据类型之一，用于表示具有嵌套关系的数据。例如，我们可

avatar 腾讯云开发者社区

终极指南：Flink SQL连接器版本管理从混乱到有序的升级之路

Apache Flink作为流处理领域的佼佼者，其SQL连接器的版本管理一直是开发者面临的核心挑战。本文将系统讲解Flink SQL连接器版本管理的最佳实践，帮助你轻松应对版本兼容性问题，实现从混乱到有序的升级之旅。## 连接器版本管理的常见痛点 😫在Flink应用开发中，连接器版本管理常常让开发者头疼不已。不同版本的连接器可能导致各种兼容性问题，例如API变更、功能差异甚至运行时错误。

avatar 腾讯云开发者社区

如何快速搭建Neon无服务器PostgreSQL：面向初学者的完整指南

Neon是一款革命性的无服务器PostgreSQL解决方案，它通过分离存储和计算层，实现了自动扩缩容、类代码式数据库分支以及零级扩展能力。本指南将帮助你从零开始搭建Neon开发环境，体验这款创新数据库的强大功能。## 准备工作：环境要求与依赖项在开始搭建Neon环境前，请确保你的系统满足以下要求：- Linux操作系统（推荐Ubuntu 20.04+或Debian 11+）- Git

avatar 腾讯云开发者社区