如何用Drizzle ORM获取单表行及关联表的一对多数据

在网络安全项目中使用如下Drizzle ORM Schema：

export const ips = createTable("ips", {
  id: uuid("id")
    .primaryKey()
    .default(sql`gen_random_uuid()`),
  ipAddress: varchar("ip_address").notNull().unique(),
  city: varchar("city", { length: 64 }).notNull(),
  region: varchar("region", { length: 64 }).notNull(),
  country: varchar("country", { length: 64 }).notNull(),
  latitude: varchar("latitude", { length: 64 }).notNull(),
  longitude: varchar("longitude", { length: 64 }).notNull(),
  timezone: varchar("timezone", { length: 64 }).notNull(),
  asn: varchar("asn", { length: 64 }).notNull(),
  isp: varchar("isp", { length: 64 }).notNull(),
  updatedAt: timestamp("updated_at").defaultNow(),
  createdAt: timestamp("created_at").defaultNow(),
});

export const ports = createTable(
  "ports",
  {
    ipId: uuid("ip_id").references(() => ips.id, { onDelete: "cascade" }),
    port: integer("port").notNull(),
    service: varchar("service", { length: 64 }).notNull(),
    rawData: text("raw_data"),
    updatedAt: timestamp("updated_at").defaultNow(),
    createdAt: timestamp("created_at").defaultNow(),
  },
  (table) => {
    return {
      pk: primaryKey({ columns: [table.ipId, table.port] }),
    };
  },
);

export const portsRelations = relations(ports, ({ one }) => ({
  ports: one(ips, {
    fields: [ports.ipId],
    references: [ips.id],
  }),
}));

已实现分页搜索函数：

export const panginatedSearchResultsSchema = z.object({
  page: z.coerce.number().default(1),
  per_page: z.coerce.number().default(10),
  sort: z.string().optional(),
  query: z.string().optional(),
  ipAddress: z.string().optional(),
  port: z.coerce.number().optional(),
  service: z.string().optional(),
});

export type GetPaginatedSearchResult = z.infer<
  typeof panginatedSearchResultsSchema
>;

export async function getPaginatedSearchQuery(input: GetPaginatedSearchResult) {
  noStore();
  const conditions = [];

  if (input.ipAddress)
    conditions.push(ilike(ips.ipAddress, `%${input.ipAddress}%`));

  if (input.port) conditions.push(eq(ports.port, input.port));

  if (input.service)
    conditions.push(ilike(ports.service, `%${input.service}%`));

  if (input.query) {
    const searchText = `%${input.query}%`;
    conditions.push(
      or(
        ilike(ips.ipAddress, searchText),
        ilike(ports.service, searchText),
        ilike(ports.rawData, searchText),
      ),
    );
  }

  const offset = ((input.page || 1) - 1) * (input.per_page || 10);
  const limit = input.per_page || 10;

  const [column, order] = (input.sort?.split(".") as [
    keyof typeof ips.$inferSelect | undefined,
    "asc" | "desc" | undefined,
  ]) ?? ["created_at", "desc"];

  const orderBy = 
    column && column in ips
      ? order === "asc"
        ? asc(ips[column])
        : desc(ips[column])
      : desc(ips.createdAt);

  const dataQuery = db
    .select({
      ipId: ips.id,
      ipAddress: ips.ipAddress,
      createdAt: ips.createdAt,
      city: ips.city,
      region: ips.region,
      country: ips.country,
      latitude: ips.latitude,
      longitude: ips.longitude,
      timezone: ips.timezone,
      asn: ips.asn,
      isp: ips.isp,
      service: ports.service,
      rawData: ports.rawData,
    })
    .from(ips)
    .leftJoin(ports, eq(ips.id, ports.ipId))
    .where(conditions.length > 0 ? and(...conditions) : undefined)
    .orderBy(orderBy)
    .limit(limit)
    .offset(offset);

  const countQuery = db
    .select({ count: count() })
    .from(ips)
    .leftJoin(ports, eq(ips.id, ports.ipId))
    .where(conditions.length > 0 ? and(...conditions) : undefined);

  const [data, countResult] = await Promise.all([
    dataQuery.execute(),
    countQuery.execute(),
  ]);

  const total = countResult[0]?.count || 0;
  const pageCount = Math.ceil(total / limit);

  return {
    data,
    total,
    pageCount,
  };
}

当前函数正常工作，但一个IP对应多个端口时会生成多条独立结果。希望搜索IP时，在单个对象中获取IP表数据及所有关联的端口表数据，且通过一次查询/事务完成，请问是否可行？

完全可行，通过Drizzle ORM的关系查询和聚合函数就能实现，同时保留原有分页、过滤逻辑的正确性。以下是具体实现步骤：

1. 补充双向关系定义 #

先给ips表添加反向关系，让Drizzle能识别IP与端口的一对多关联：

// 新增ips表的反向关系
export const ipsRelations = relations(ips, ({ many }) => ({
  ports: many(ports),
}));

2. 修改查询逻辑，实现嵌套数据返回 #

使用数据库聚合函数（以PostgreSQL的json_agg为例）将每个IP的端口数据合并为数组，同时调整过滤和计数逻辑，确保分页针对唯一IP：

修改后的完整函数：

export async function getPaginatedSearchQuery(input: GetPaginatedSearchResult) {
  noStore();
  const ipConditions = [];
  const portConditions = [];

  // 拆分IP和端口的过滤条件
  if (input.ipAddress) {
    ipConditions.push(ilike(ips.ipAddress, `%${input.ipAddress}%`));
  }

  if (input.port) {
    portConditions.push(eq(ports.port, input.port));
  }

  if (input.service) {
    portConditions.push(ilike(ports.service, `%${input.service}%`));
  }

  // 处理全局搜索：匹配IP或关联的端口数据
  if (input.query) {
    const searchText = `%${input.query}%`;
    ipConditions.push(
      or(
        ilike(ips.ipAddress, searchText),
        exists(
          db.select().from(ports).where(
            and(
              eq(ports.ipId, ips.id),
              or(
                ilike(ports.service, searchText),
                ilike(ports.rawData, searchText)
              )
            )
          )
        )
      )
    );
  }

  const offset = ((input.page || 1) - 1) * (input.per_page || 10);
  const limit = input.per_page || 10;

  const [column, order] = (input.sort?.split(".") as [
    keyof typeof ips.$inferSelect | undefined,
    "asc" | "desc" | undefined,
  ]) ?? ["created_at", "desc"];

  const orderBy = 
    column && column in ips
      ? order === "asc"
        ? asc(ips[column])
        : desc(ips[column])
      : desc(ips.createdAt);

  // 主查询：聚合端口数据到IP对象中
  const dataQuery = db
    .select({
      ipId: ips.id,
      ipAddress: ips.ipAddress,
      createdAt: ips.createdAt,
      city: ips.city,
      region: ips.region,
      country: ips.country,
      latitude: ips.latitude,
      longitude: ips.longitude,
      timezone: ips.timezone,
      asn: ips.asn,
      isp: ips.isp,
      // 聚合端口数据为JSON数组，过滤掉无端口的null值
      ports: sql`json_agg(json_build_object(
        'port', ${ports.port},
        'service', ${ports.service},
        'rawData', ${ports.rawData},
        'updatedAt', ${ports.updatedAt},
        'createdAt', ${ports.createdAt}
      )) filter (where ${ports.ipId} is not null)`
    })
    .from(ips)
    .leftJoin(ports, eq(ips.id, ports.ipId))
    .where(
      and(
        ...ipConditions,
        // 若有端口过滤条件，只保留符合条件的IP
        portConditions.length > 0 ? exists(
          db.select().from(ports).where(
            and(eq(ports.ipId, ips.id), ...portConditions)
          )
        ) : undefined
      )
    )
    .groupBy(ips.id)
    .orderBy(orderBy)
    .limit(limit)
    .offset(offset);

  // 计数查询：统计符合条件的唯一IP数量
  const countQuery = db
    .select({ count: countDistinct(ips.id) })
    .from(ips)
    .leftJoin(ports, eq(ips.id, ports.ipId))
    .where(
      and(
        ...ipConditions,
        portConditions.length > 0 ? exists(
          db.select().from(ports).where(
            and(eq(ports.ipId, ips.id), ...portConditions)
          )
        ) : undefined
      )
    );

  const [data, countResult] = await Promise.all([
    dataQuery.execute(),
    countQuery.execute(),
  ]);

  // 格式化端口数据（PostgreSQL返回的JSON数组无需手动解析，若为字符串则需JSON.parse）
  const formattedData = data.map(item => ({
    ...item,
    ports: item.ports || []
  }));

  const total = countResult[0]?.count || 0;
  const pageCount = Math.ceil(total / limit);

  return {
    data: formattedData,
    total,
    pageCount,
  };
}

关键细节说明 #

• 使用json_agg聚合端口数据，通过filter确保无端口的IP返回空数组而非null
• 端口过滤条件通过exists子查询实现，保证只返回关联端口符合要求的IP
• 计数使用countDistinct(ips.id)，避免因端口数量导致计数重复
• 保持原有分页、排序和全局搜索逻辑，同时实现IP与端口数据的嵌套结构

简化方案（Drizzle Relations API） #

若你的Drizzle版本支持关系查询的with方法，可直接使用内置关联语法，代码更简洁：

const dataQuery = db
  .select()
  .from(ips)
  .with({ ports: ipsRelations.ports })
  .where(/* 合并后的过滤条件 */)
  .orderBy(orderBy)
  .limit(limit)
  .offset(offset);

注意：此方式分页仍针对IP，端口过滤条件同样需要用exists子查询保证正确性。

内容的提问来源于stack exchange，提问作者TheLawlessDev