序文

显贵的读者朋友们，在上一章之末，咱们现已成功完结了应用程序装备信息与模型界说部分的工作。在此，咱们即将深入探讨如何运用SQLAlchemy工具包来快捷地创立数据库表，同时，咱们也将演示用户信息的常见CURD（Create，Update，Read，Delete）操作以及短链接信息表的CURD封装流程。

创立表的诀窍：SQLAlchemy

在同步引擎环境下，假设您选用的是create_engine函数并赋值为sqlite:///，那么只需简单地调用Base.metadata.create_all(engine, checkfirst=True)即可轻松完结数据表的创立。这儿的checkfirst参数设置为True，意味着首次初始化时才会创立表；若表已存在，再次执行此操作便不会发生任何影响。

而在异步引擎环境下，enginwe = create_async_engine('sqlite:///')，您可以按照以下步骤进行表的创立：

async def init_create_table():
  async with async_engine.begin() as conn:
    await conn.run_sync(Base.metadata.drop_all)
    await conn.run_sync(Base.metadata.create_all)

咱们将表的创立进程集成到app发动事件回调中，修正代码如下：

@app.on_event("startup")
async def startup_event():
  from db.database import async_engine, Base
  from models.model import User,ShortUrl
  async def init_create_table():
    async with async_engine.begin() as conn:
      await conn.run_sync(Base.metadata.drop_all)
      await conn.run_sync(Base.metadata.create_all)
  await init_create_table()

在服务发动事件on_event(“startup”)回调中进行了数据库表的创立，经过async_engine.begin() as conn获取了一个衔接数据库的会话对象，然后执行数据库表的删去和从头创立的操作，其中，Base.metadata.drop_all用于删去表，Base.metadata.create_all用于创立表。对表进行删去需谨慎 生成表时需求留意，要导入具体的模型类，如from models.model import User，ShortUrl。假如不导入，则执行创立表的操作时会因无法找到要生成的表模型而犯错。 当发动服务之后，就可以看到在当前项目目录下生成了数据库文件。

用户信息CRUD

先来看完成代码，services/user.py

from sqlalchemy import select, update, delete,func
from sqlalchemy.ext.asyncio import AsyncSession
​
from models.model import User
from db.database import async_engine, Base
​
​
class UserServeries:
​
​
  @staticmethod
  async def get_user(async_session: AsyncSession, user_id: int):
    result = await async_session.execute(select(User).where(User.id == user_id))
    return result.scalars().first()
​
  @staticmethod
  async def get_user_by_name(async_session: AsyncSession, username: str) -> User:
    result = await async_session.execute(select(User).where(User.username == username))
    return result.scalars().first()
​
  @staticmethod
  async def get_users(async_session: AsyncSession):
    result = await async_session.execute(select(User).order_by(User.id))
    return result.scalars().fetchall()
​
  @staticmethod
  async def create_user(async_session: AsyncSession, **kwargs):
    new_user = User(**kwargs)
    async_session.add(new_user)
    await async_session.commit()
    return new_user
​
  @staticmethod
  async def update_user(async_session: AsyncSession, user_id: int, **kwargs):
    response = update(User).where(User.id == user_id)
    result = await async_session.execute(response.values(**kwargs))
    await async_session.commit()
    return result
​
  @staticmethod
  async def delete_user(async_session: AsyncSession, user_id: int):
    response = await async_session.execute(delete(User).where(User.id == user_id))
    await async_session.commit()
    return response

UserServeries类统一管理user表的操作，在该类中界说了与user表的增、删、改、查等操作对应的静态协程办法。在所有的静态协程办法中，第一个参数统一依赖于AsyncSession异步的会话对象。

  • create_user()办法用于创立User用户信息。
  • get_user()、get_user_by_name()、get_users()等办法用于依据不同的条件查询User用户信息。
  • update_user()办法用于依据用户ID修正User用户信息。
  • delete_user()办法用于依据用户ID删去User用户记载。

短链信息CURD

仍是先看完成代码，services/short.py

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from models.model import ShortUrl
from db.database import async_engine, Base
from typing import List
from schemas import SingleShortUrlCreate
​
class ShortServeries:
​
  
  @staticmethod
  async def get_short_url(async_session: AsyncSession, short_tag: str)->ShortUrl:
    result = await async_session.execute(select(ShortUrl).where(ShortUrl.short_tag == short_tag))
    return result.scalars().first()
​
  @staticmethod
  async def create_short_url(async_session: AsyncSession, **kwargs):
    new_short_url = ShortUrl(**kwargs)
    async_session.add(new_short_url)
    await async_session.commit()
    return new_short_url
​
  @staticmethod
  async def update_short_url(async_session: AsyncSession, short_url_id: int, **kwargs):
    response = update(ShortUrl).where(ShortUrl.id == short_url_id)
    result = await async_session.execute(response.values(**kwargs))
    await async_session.commit()
    return result
​
  @staticmethod
  async def delete_short_url(async_session: AsyncSession, short_url_id: int):
    response = await async_session.execute(delete(ShortUrl).where(ShortUrl.id == short_url_id))
    await async_session.commit()
    return response
​
​
  @staticmethod
  async def create_batch_short_url(async_session: AsyncSession, short_urls:List[SingleShortUrlCreate]):
    short_urls= [ShortUrl(**item.dict()) for item in short_urls]
    async_session.add_all(short_urls)
    await async_session.commit()
    return short_urls

ShortServeries类来统一管理短链信息表的操作，该类中界说了与短链信息表的增、删、改、查等操作对应的静态协程办法，这些静态协程办法包含：

  • get_short_url()办法用于依据short_tag查询短链记载。
  • create_short_url()办法用于创立单一的短链记载。
  • update_short_url()办法用于依据短链记载的ID更新短链信息。
  • delete_short_url()办法用于依据短链记载的ID删去短链记载。
  • create_batch_short_url()办法用于批量增加多条短链记载。

后续展望

在这篇文章中，主要是讲了怎样用SQLAlchemy这个强大工具来创立各种表，并且还有具体讲解了怎样操作那些用户信息表和短链信息表。

敬请期待：下一篇文章将介绍接口的界说以及路由装备。在此，咱们诚挚邀请各位朋友继续重视此系列文章，若有所得，恳请您给予点赞和保藏以示鼓舞，不胜感激！