跳至内容

查询操作

为了更全面的文档，本页记录了在 prisma 客户端上进行查询操作，例如创建、查找、更新和删除记录。

模式

示例使用以下 prisma 模式模型 

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

model User {
  id      String   @id @default(cuid())
  name    String
  points  Float    @default(0)
  meta    Json     @default("{}")
  number  Decimal  @default(0)
  emails  String[]
  posts   Post[]
  profile Profile?
}

model Profile {
  id      Int     @id @default(autoincrement())
  user    User    @relation(fields: [user_id], references: [id])
  user_id String  @unique
  bio     String
  image   Bytes?
  city    String?
  country String?
  views   Int     @default(0)
}

model Post {
  id          String     @id @default(cuid())
  created_at  DateTime   @default(now())
  updated_at  DateTime   @updatedAt
  title       String
  published   Boolean
  views       Int        @default(0)
  description String?
  author      User?      @relation(fields: [author_id], references: [id])
  author_id   String?
  categories  Category[]
}

model Category {
  id    Int    @id @default(autoincrement())
  posts Post[]
  name  String
}

创建

单个记录

user = await db.user.create(
    data={
        'name': 'Robert',
    },
)

多条记录

警告

create_many 不适用于 SQLite

users = await db.user.create_many(
    data=[
        {'name': 'Tegan'},
        {'name': 'Alfie'},
        {'name': 'Robert'},
    ]
)
users = await db.user.create_many(
    data=[
        {'id': 'abc', 'name': 'Tegan'},
        {'id': 'def', 'name': 'Alfie'},
        {'id': 'ghi', 'name': 'Robert'},
    ],
    skip_duplicates=True,
)

关联记录

user = await db.user.create(
    data={
        'name': 'Robert',
        'profile': {
            'create': {
                'bio': 'My very cool bio!',
            }
        }
    }
)
user = await db.user.create(
    data={
        'name': 'Robert',
        'posts': {
            'create': [
                {
                    'title': 'My first post!',
                    'published': True,
                },
                {
                    'title': 'My draft post!',
                    'published': False,
                },
            ]
        }
    }
)

查找

唯一记录

这里有两种不同的方法可以用来查找唯一的记录。使用哪种方法取决于查询周围的上下文

  • 如果预期记录不存在于数据库中，请使用 find_unique
  • 如果意外记录不存在于数据库中，请使用 find_unique_or_raise

user = await db.user.find_unique(
    where={
        'id': '1',
    }
)
user = await db.user.find_unique_or_raise(
    where={
        'id': '2',
    },
    include={
        'posts': True,
    },
)

单个记录

这里有两种不同的方法可以用来查找单个记录。使用哪种方法取决于查询周围的上下文

  • 如果预期记录不存在于数据库中，请使用 find_first
  • 如果意外记录不存在于数据库中，请使用 find_first_or_raise

post = await db.post.find_first(
    where={
        'title': {'contains': 'Post'},
    },
)
post = await db.post.find_first_or_raise(
    where={
        'title': {'contains': 'Post'},
    },
)
post = await db.post.find_first(
    skip=2,
    where={
        'title': {
            'contains': 'Post'
        },
    },
    cursor={
        'id': 'abcd',
    },
    include={
        'author': True,
    },
    order={
        'id': 'asc',
    }
)

多条记录

posts = await db.post.find_many(
    where={
        'published': True,
    },
)
posts = await db.post.find_many(
    take=5,
    skip=1,
    where={
        'published': True,
    },
    cursor={
        'id': 'desc',
    },
    include={
        'categories': True,
    },
    order={
        'id': 'desc',
    }
)

不同记录

以下查询将查找所有具有不同 city 字段的 Profile 记录。

profiles = await db.profiles.find_many(
    distinct=['city'],
)
# [
#  { city: 'Paris' },
#  { city: 'Lyon' },
# ]

您还可以按不同的组合进行过滤，例如以下查询将返回所有具有不同 city 和 country 组合的记录。

profiles = await db.profiles.find_many(
    distinct=['city', 'country'],
)
# [
#  { city: 'Paris', country: 'France' },
#  { city: 'Paris', country: 'Denmark' },
#  { city: 'Lyon', country: 'France' },
# ]

目前您只能在 find_many() 和 find_first() 查询中使用 distinct。

按关联字段过滤

在过滤器中，您可以查询所有通常查询的内容，就像它是在关联字段上进行 find_first() 调用一样，例如

post = await db.post.find_first(
    where={
        'author': {
            'is': {
                'name': 'Robert',
            },
        },
    },
)
user = await db.user.find_first(
    where={
        'name': 'Robert',
    },
)

一对一

post = await db.post.find_first(
    where={
        'author': {
            'is': {
                'name': 'Robert',
            },
            'is_not': {
                'name': 'Tegan',
            },
        },
    },
)

一对多

排除
post = await db.post.find_first(
    where={
        'categories': {
            'none': {
                'name': 'Exclude Category',
            },
        },
    },
)
至少一个
post = await db.post.find_first(
    where={
        'categories': {
            'some': {
                'name': {
                    'contains': 'Special',
                },
            },
        },
    },
)
每个
post = await db.post.find_first(
    where={
        'categories': {
            'every': {
                'name': {
                    'contains': 'Category',
                },
            },
        },
    },
)

按字段值过滤

注意

过滤字段的示例只是为了展示可能的参数，所有一起传递的参数将导致无效的查询或找不到任何记录。

字符串字段

警告

不区分大小写的过滤仅适用于 PostgreSQL 和 MongoDB

post = await db.post.find_first(
    where={
        'description': 'Must be exact match',
        # or
        'description': {
            'equals': 'example_string',
            'not_in': ['ignore_string_1', 'ignore_string_2'],
            'lt': 'z',
            'lte': 'y',
            'gt': 'a',
            'gte': 'b',
            'contains': 'string must be present',
            'startswith': 'must start with string',
            'endswith': 'must end with string',
            'in': ['find_string_1', 'find_string_2'],
            'mode': 'insensitive',
            'not': {
                # recursive type
                'contains': 'string must not be present',
                'mode': 'default',
                ...
            },
        },
    },
)

整数字段

post = await db.post.find_first(
    where={
        'views': 10,
        # or
        'views': {
            'equals': 1,
            'in': [1, 2, 3],
            'not_in': [4, 5, 6],
            'lt': 10,
            'lte': 9,
            'gt': 0,
            'gte': 1,
            'not': {
                # recursive type
                'gt': 10,
                ...
            },
        },
    },
)

浮点字段

user = await db.user.find_first(
    where={
        'points': 10.0,
        # or
        'points': {
            'equals': 10.0,
            'in': [1.2, 1.3, 1.4],
            'not_in': [4.7, 53.4, 6.8],
            'lt': 100.5,
            'lte': 9.9,
            'gt': 0.0,
            'gte': 1.2,
            'not': {
                # recursive type
                'gt': 10.0,
                ...
            },
        },
    },
)

日期时间字段

from datetime import datetime

post = await db.post.find_first(
    where={
        'updated_at': datetime.now(),
        # or
        'updated_at': {
            'equals': datetime.now(),
            'not_in': [datetime.now(), datetime.utcnow()],
            'lt': datetime.now(),
            'lte': datetime.now(),
            'gt': datetime.now(),
            'gte': datetime.now(),
            'in': [datetime.now(), datetime.utcnow()],
            'not': {
                # recursive type
                'equals': datetime.now(),
                ...
            },
        },
    },
)

布尔字段

post = await db.post.find_first(
    where={
        'published': True,
        # or
        'published': {
            'equals': True,
            'not': False,
        },
    },
)

Json 字段

注意

Json 字段必须完全匹配。

警告

Json 字段在 SQLite 上不受支持

from prisma import Json

user = await db.user.find_first(
    where={
        'meta': Json({'country': 'Scotland'})
        # or
        'meta': {
            'equals': Json.keys(country='Scotland'),
            'not': Json(['foo']),
        }
    }
)

字节字段

注意

字节字段被编码为和从 Base64 编码

from prisma import Base64

profile = await db.profile.find_first(
    where={
        'image': Base64.encode(b'my binary data'),
        # or
        'image': {
            'equals': Base64.encode(b'my binary data'),
            'in': [Base64.encode(b'my binary data')],
            'not_in': [Base64.encode(b'my other binary data')],
            'not': Base64(b'WW91IGZvdW5kIGFuIGVhc3RlciBlZ2chIExldCBAUm9iZXJ0Q3JhaWdpZSBrbm93IDop'),
        },
    },
)

十进制字段

from decimal import Decimal

user = await db.user.find_first(
    where={
        'number': Decimal(1),
        # or
        'number': {
            'equals': Decimal('1.23823923283'),
            'in': [Decimal('1.3'), Decimal('5.6')],
            'not_in': [Decimal(10), Decimal(20)],
            'gte': Decimal(5),
            'gt': Decimal(11),
            'lt': Decimal(4),
            'lte': Decimal(3),
            'not': Decimal('123456.28'),
            # or
            'not': {
                # recursive type
                ...
            }
        },
    },
)

列表字段

警告

标量列表字段仅在 PostgreSQL、CockroachDB 和 MongoDB 上受支持

每个标量类型也可以定义为列表，例如

user = await db.user.find_first(
    where={
        'emails': {
            # only one of the following fields is allowed at the same time
            'has': 'robert@craigie.dev',
            'has_every': ['email1', 'email2'],
            'has_some': ['email3', 'email4'],
            'is_empty': True,
        },
    },
)

组合参数

以上提到的所有过滤器都可以使用 AND、NOT 和 OR 与其他过滤器组合。

AND

以下查询将返回标题包含单词 prisma 和 test 的第一个帖子。

post = await db.post.find_first(
    where={
        'AND': [
            {
                'title': {
                    'contains': 'prisma',
                },
            },
            {
                'title': {
                    'contains': 'test',
                },
            },
        ],
    },
)

OR

以下查询将返回标题包含单词 prisma 或已发布的第一个帖子。

post = await db.post.find_first(
    where={
        'OR': [
            {
                'title': {
                    'contains': 'prisma',
                },
            },
            {
                'published': True,
            },
        ],
    },
)

NOT

以下查询将返回标题不是 My test post 的第一个帖子。

post = await db.post.find_first(
    where={
        'NOT' [
            {
                'title': 'My test post',
            },
        ],
    },
)

删除

唯一记录

post = await db.post.delete(
    where={
        'id': 'cksc9m7un0028f08zwycxtjr1',
    },
)
post = await db.post.delete(
    where={
        'id': 'cksc9m1vu0021f08zq0066pnz',
    },
    include={
        'categories': True,
    }
)

多条记录

total = await db.post.delete_many(
    where={
        'published': False,
    }
)

更新

唯一记录

post = await db.post.update(
    where={
        'id': 'cksc9lp7w0014f08zdkz0mdnn',
    },
    data={
        'views': {
            'increment': 1,
        }
    },
    include={
        'categories': True,
    }
)

多条记录

total = await db.post.update_many(
    where={
        'published': False
    },
    data={
        'views': 0,
    },
)

未找到时创建

post = await db.post.upsert(
    where={
        'id': 'cksc9ld4z0007f08z7obo806s',
    },
    data={
        'create': {
            'title': 'This post was created!',
            'published': False,
        },
        'update': {
            'title': 'This post was updated',
            'published': True,
        },
    },
    include={
        'categories': True,
    }
)

更新原子字段

如果字段是 int 或 float 类型，则可以对其进行原子更新，即可以在不知道先前值的情况下应用数学运算。

整数字段

post = await db.post.update(
    where={
        'id': 'abc',
    },
    data={
        'views': 1,
        # or
        'views': {
            'set': 5,
            'increment': 1,
            'decrement': 2,
            'multiply': 5,
            'divide': 10,
        },
    },
)

浮点字段

user = await db.user.update(
    where={
        'id': 'abc',
    },
    data={
        'points': 1.0,
        # or
        'points': {
            'set': 1.0,
            'increment': 1.5,
            'decrement': 0.5,
            'multiply': 2.5,
            'divide': 3.0,
        },
    },
)

更新列表字段

警告

标量列表字段仅在 PostgreSQL、CockroachDB 和 MongoDB 上受支持

警告

CockroachDB 不支持 push 操作

user = await db.user.update(
    where={
        'id': 'cksc9lp7w0014f08zdkz0mdnn',
    },
    data={
        'email': {
            'set': ['robert@craigie.dev', 'robert@example.com'],
            # or
            'push': ['robert@example.com'],
        },
    }
)

聚合

统计记录

total = await db.post.count(
    where={
        'published': True,
    },
)
total = await db.post.count(
    take=10,
    skip=1,
    where={
        'published': True,
    },
    cursor={
        'id': 'cksca3xm80035f08zjonuubik',
    },
)

分组记录

警告

您一次只能按一个字段排序，但是这 无法用 python 类型表示

results = await db.profile.group_by(['country'])
# [
#   {'country': 'Denmark'},
#   {'country': 'Scotland'},
# ]

results = await db.profile.group_by(['country'], count=True)
# [
#   {'country': 'Denmark', '_count': {'_all': 20}},
#   {'country': 'Scotland', '_count': {'_all': 1}},
# ]

results = await db.profile.group_by(
    by=['country', 'city'],
    count={
        '_all': True,
        'city': True,
    },
    sum={
        'views': True,
    },
    order={
        'country': 'desc',
    },
    having={
        'views': {
            '_avg': {
                'gt': 200,
            },
        },
    },
)
# [
#   {
#       'country': 'Scotland',
#       'city': 'Edinburgh',
#       '_sum': {'views': 250},
#       '_count': {'_all': 1, 'city': 1}
#   },
#   {
#       'country': 'Denmark',
#       'city': None,
#       '_sum': {'views': 6000},
#       '_count': {'_all': 12, 'city': 0}
#   },
#   {
#       'country': 'Denmark',
#       'city': 'Copenhagen',
#       '_sum': {'views': 8000},
#       '_count': {'_all': 8, 'city': 8}
#   },
# ]

批量写入查询

async with db.batch_() as batcher:
    batcher.user.create({'name': 'Robert'})
    batcher.user.create({'name': 'Tegan'})

原始查询

注意

SQL 查询直接发送到数据库，因此您必须使用特定数据库提供程序的语法

警告

原始查询结果是原始字典，除非指定了 model 参数

写入查询

total = await db.execute_raw(
    '''
    SELECT *
    FROM User
    WHERE User.id = ?
    ''',
    'cksca3xm80035f08zjonuubik'
)

选择多条记录

posts = await db.query_raw(
    '''
    SELECT *
    FROM Post
    WHERE Post.published IS TRUE
    '''
)

类型安全

from prisma.models import Post

posts = await db.query_raw(
    '''
    SELECT *
    FROM Post
    WHERE Post.published IS TRUE
    ''',
    model=Post,
)

选择单个记录

post = await db.query_first(
    '''
    SELECT *
    FROM Post
    WHERE Post.published IS TRUE
    LIMIT 1
    '''
)

类型安全

from prisma.models import Post

post = await db.query_first(
    '''
    SELECT *
    FROM Post
    WHERE Post.views > 50
    LIMIT 1
    ''',
    model=Post
)