Skip to main content

基于SQLModel封装的专用于FastAPI处理MySQL数据库业务的框架

Project description

fastzdp_sqlmodel

基于SQLModel封装的专用于FastAPI处理MySQL数据库业务的框架

Github开源地址:https://github.com/zhangdapeng520/fastzdp_sqlmodel

安装

pip install fastzdp_sqlmodel

使用教程

创建表

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 删除所有表
SQLModel.metadata.drop_all(engine)

# 创建所有表
SQLModel.metadata.create_all(engine)

添加数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)

查询所有数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询所有数据
users = fsqlmodel.get_all(engine, User)
print(users)

根据字典查询数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="张三", age=24)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询所有数据
query = {"name": "张三"}
users = fsqlmodel.get_by_dict(engine, User, query)
print(users)

逻辑 or 查询

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="张三", age=33)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询年龄等于23或者小于10的用户
query = {"age": {">": 23, "<": 10}}
users = fsqlmodel.get_by_dict_or(engine, User, query)
print(users)

逻辑 and 查询

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="张三", age=33)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询年龄等于20或者小于30的用户
query = {"age": {">": 20, "<": 30}}
users = fsqlmodel.get_by_dict_and(engine, User, query)
print(users)

根据ID查询

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)
u = User(name="张三", age=33)
fsqlmodel.add(engine, u)
u = User(name="李四", age=24)
fsqlmodel.add(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

# 查询id为11的数据
u = fsqlmodel.get(engine, User, 11)
print(u)

修改数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

# 修改
update_dict = {"name": "张三333"}
fsqlmodel.update(engine, u, update_dict)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

根据ID修改数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

# 修改
update_dict = {"name": "张三333"}
fsqlmodel.update_id(engine, User, 1, update_dict)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

删除数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

# 删除
fsqlmodel.delete(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

根据ID删除数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
u = User(name="张三", age=23)
fsqlmodel.add(engine, u)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

# 删除
fsqlmodel.delete_id(engine, User, 1)

# 查询id为1的数据
u = fsqlmodel.get(engine, User, 1)
print(u)

分页查询数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fsqlmodel


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fsqlmodel.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fsqlmodel.init_table(engine)

# 创建数据
fsqlmodel.add(engine, User(name="张三1", age=23))
fsqlmodel.add(engine, User(name="张三2", age=23))
fsqlmodel.add(engine, User(name="张三3", age=23))

# 分页查询
result = fsqlmodel.get_page(engine, User, page=1, size=2)
print(result)

获取第一条数据

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fs


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fs.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fs.init_table(engine)

# 创建数据
fs.add(engine, User(name="张三1", age=23))
fs.add(engine, User(name="张三2", age=23))
fs.add(engine, User(name="张三3", age=23))

# 查询第一条数据
u = fs.get_first(engine, User, {"name": "张三1"})
print(u)

判断数据是否存在

from typing import Optional
from sqlmodel import Field, SQLModel
import fastzdp_sqlmodel as fs


class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    age: Optional[int] = None


# 创建数据库引擎
engine = fs.get_engine(database="fastzdp_sqlmodel")

# 初始化表格
fs.init_table(engine)

# 创建数据
fs.add(engine, User(name="张三1", age=23))
fs.add(engine, User(name="张三2", age=23))
fs.add(engine, User(name="张三3", age=23))

# 判断是否存在
print(fs.is_exists(engine, User, {"name": "张三1"}))
print(fs.is_exists(engine, User, {"name": "张三11"}))

版本历史

v0.1.0

  • 常见的增删改查功能

v0.1.1

  • get_page 分页查询
  • get_first 获取第一条数据
  • is_exists 判断数据是否存在

v0.1.2

  • get_page 根据ID降序

v0.1.3

  • get_page 支持等值查询

v0.1.4

  • 新增get_sum求和方法

v0.1.5

  • 支持Python3.7版本

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

fastzdp_sqlmodel-0.1.5.tar.gz (9.1 kB view details)

Uploaded Source

Built Distribution

fastzdp_sqlmodel-0.1.5-py3-none-any.whl (11.8 kB view details)

Uploaded Python 3

File details

Details for the file fastzdp_sqlmodel-0.1.5.tar.gz.

File metadata

  • Download URL: fastzdp_sqlmodel-0.1.5.tar.gz
  • Upload date:
  • Size: 9.1 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.8.3 CPython/3.12.4 Windows/11

File hashes

Hashes for fastzdp_sqlmodel-0.1.5.tar.gz
Algorithm Hash digest
SHA256 dbf2129c53c9e780592c04387f989f40992e13cc6713f47bd9bdbff70c63f7cd
MD5 982b60501f185eb1b3919961f17ab488
BLAKE2b-256 70b95f6e067b8e5b80d13958ea8be25949b911922cb762122465e96e2947d9d6

See more details on using hashes here.

File details

Details for the file fastzdp_sqlmodel-0.1.5-py3-none-any.whl.

File metadata

File hashes

Hashes for fastzdp_sqlmodel-0.1.5-py3-none-any.whl
Algorithm Hash digest
SHA256 36c6ecfc17835a56a6b65304de8d1b09a0c0ce3f1ce786b53de32002765a0fd8
MD5 4f3aeac48e47712526ffd3d4f24e1b0b
BLAKE2b-256 8d332e76eb49e14f7b54ae501474b497bab51d02b65f2de0cd20d7f8d8d1af46

See more details on using hashes here.

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page