Building a Data Pipeline: Writing MySQL Query Results to Feishu Bitable Tables

Overview

Consider a practical scenario where a company launches a new product and needs to perform customer follow-ups with users who placed orders. The业务 team requires specific user information (nickname, gender, phone, email) combined with order details (order ID, purchase timestamp, order amount) to facilitate outreach efforts.

This guide demonstrates how to build a complete pipeline that queries MySQL and writes the results directly to a Feishu bitable for collaborative team access.

Pipeline Architecture

The data synchronization pipeline consists of five distinct phases:

  • Phase 1: Target Analysis - Already defined in the scenario above
  • Phase 2: Table Creation - Create the bitable structure via API
  • Phase 3: Data Extraction - Connect to MySQL and retrieve query results
  • Phase 4: Data Transformation - Convert query results to Feishu batch create API format
  • Phase 5: Data Insertion - Write the formatted data to the bitable

The following sections elaborate on each phase starting from Phase 2.

Creating the Bitable Structure

Before creating the table, field types must be determined based on bitable capabilities.

The first column typically serves as a unique identifier. Since each user places only one order, the user ID field can serve this purpose as it maintains a one-to-one relationship with order records. Alternatively, an auto-incrementing ID column could be added as the primary key.

Text fields like nickname, phone, and order ID offer basic filtering (exact and pattern matching). For more advanced filtering capabiilties, consider converting to single-select fields where applicable. For example, gender and city fields work well as single-select types.

Important Constraint: Single-select fields are limited to 5,000 options maximum. Exceeding this limit requires using multi-line text fields instead.

The final field configuration is summarized below:

Field Name Field Type type ui_type API Data Format
UserID Auto-increment 2 Number Integer
Nickname Multi-line Text 1 Text String
Gender Single Select 3 SingleSelect String
Phone Phone 13 Phone String
City Single Select 3 SingleSelect String
OrderID Multi-line Text 1 Text String
PurchaseTimestamp DateTime 5 DateTime Integer (epoch milliseconds)
OrderAmount Number 2 Number Float

The API request body structuer includes property configurations:

  • UserID uses {"formatter": "0"} to display as integer
  • OrderAmount uses {"formatter": "0.00"} for two decimal places
  • PurchaseTimestamp uses {"date_formatter": "yyyy/MM/dd HH:mm", "auto_fill": false} for specific format without auto-population
{
    "table": {
        "name": "Product Customer List",
        "default_view_name": "All Records View",
        "fields": [
            {"field_name": "UserID", "type": 2, "ui_type": "Number", "property": {"formatter": "0"}},
            {"field_name": "Nickname", "type": 1, "ui_type": "Text"},
            {"field_name": "Gender", "type": 3, "ui_type": "SingleSelect"},
            {"field_name": "Phone", "type": 13, "ui_type": "Phone"},
            {"field_name": "City", "type": 3, "ui_type": "SingleSelect"},
            {"field_name": "OrderID", "type": 1, "ui_type": "Text"},
            {"field_name": "PurchaseTimestamp", "type": 5, "ui_type": "DateTime", "property": {"date_formatter": "yyyy/MM/dd HH:mm", "auto_fill": false}},
            {"field_name": "OrderAmount", "type": 2, "ui_type": "Number", "property": {"formatter": "0.00"}}
        ]
    }
}

Execute the table creation via API. Note that Python requires boolean values as True/False rather than true/false:

import requests
import json

def create_bitable(app_token, table_config, access_token):
    endpoint = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables"
    
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {access_token}'
    }
    
    response = requests.post(endpoint, headers=headers, json=table_config)
    result = response.json()
    
    if result.get('code') == 0:
        table_id = result['data']['table_id']
        print(f"Table created successfully: {table_config['table']['name']} (ID: {table_id})")
        return table_id
    else:
        raise RuntimeError(f"Table creation failed: {result.get('msg')}")


def fetch_tenant_token(app_id, app_secret):
    auth_url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    
    payload = {
        "app_id": app_id,
        "app_secret": app_secret
    }
    
    response = requests.post(auth_url, json=payload)
    token = response.json().get('tenant_access_token')
    print(f"Authentication successful")
    return token


def main():
    app_id = 'APP_ID_VALUE'
    app_secret = 'APP_SECRET_VALUE'
    app_token = 'APP_TOKEN_VALUE'
    
    access_token = fetch_tenant_token(app_id, app_secret)
    
    table_config = {
        "table": {
            "name": "Product Customer List",
            "default_view_name": "All Records View",
            "fields": [
                {"field_name": "UserID", "type": 2, "ui_type": "Number", "property": {"formatter": "0"}},
                {"field_name": "Nickname", "type": 1, "ui_type": "Text"},
                {"field_name": "Gender", "type": 3, "ui_type": "SingleSelect"},
                {"field_name": "Phone", "type": 13, "ui_type": "Phone"},
                {"field_name": "City", "type": 3, "ui_type": "SingleSelect"},
                {"field_name": "OrderID", "type": 1, "ui_type": "Text"},
                {"field_name": "PurchaseTimestamp", "type": 5, "ui_type": "DateTime", "property": {"date_formatter": "yyyy/MM/dd HH:mm", "auto_fill": False}},
                {"field_name": "OrderAmount", "type": 2, "ui_type": "Number", "property": {"formatter": "0.00"}}
            ]
        }
    }
    
    table_id = create_bitable(app_token, table_config, access_token)


if __name__ == '__main__':
    main()

Extracting and Processing MySQL Data

Combining data extraction with preprocessing is efficient because Pandas handling depends significantly on the source data format. If the SQL query already transforms data types appropriately, Pandas requires minimal additional processing.

Database Setup

Assume two source tables: a users table and an orders table.

The users table contains: user_id (primary key), mobile, nickname, gender, province, city

The orders table contains: order_id (primary key), user_id (foreign key), product_id, paid_timestamp, amount

Create the tables and insert test data:

-- Database creation
CREATE DATABASE IF NOT EXISTS product_data CHARACTER SET utf8mb4;
USE product_data;

-- Users table
CREATE TABLE users (
    user_id BIGINT AUTO_INCREMENT PRIMARY KEY,
    mobile VARCHAR(20) NOT NULL,
    nickname VARCHAR(50),
    gender TINYINT DEFAULT 2 COMMENT '0=female, 1=male, 2=unknown',
    province VARCHAR(50),
    city VARCHAR(50),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Orders table
CREATE TABLE orders (
    order_id BIGINT AUTO_INCREMENT PRIMARY KEY,
    user_id BIGINT NOT NULL,
    product_id VARCHAR(50) NOT NULL,
    paid_timestamp BIGINT NOT NULL COMMENT 'Unix timestamp',
    amount BIGINT NOT NULL COMMENT 'Amount in cents',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (user_id) REFERENCES users(user_id)
);

-- Sample data
INSERT INTO users (mobile, nickname, gender, province, city) VALUES
('13511112222', 'alice', 0, 'Beijing', 'Beijing'),
('13622223333', 'bob', 1, 'Shanghai', 'Shanghai'),
('13733334444', 'charlie', 2, 'Guangdong', 'Guangzhou'),
('13844445555', 'david', 0, 'Guangdong', 'Shenzhen');

INSERT INTO orders (user_id, product_id, paid_timestamp, amount) VALUES
(1, 'PROD001', 1704095836, 99900),
(2, 'PROD001', 1704195836, 99900),
(3, 'PROD001', 1704295836, 99900),
(4, 'PROD002', 1704395836, 89900);

Filter for product_id equal to PROD001 only.

Native SQL Transformation Approach

Transforming data directly in the SQL query is the preferred method as it simplifies downstream processing. The key transformations include converting timestamps to milliseconds and aliasing columns to match bitable field names:

SELECT 
    u.user_id AS "UserID",
    u.nickname AS "Nickname",
    CASE u.gender 
        WHEN 0 THEN 'Female' 
        WHEN 1 THEN 'Male' 
        ELSE 'Unknown' 
    END AS "Gender",
    u.mobile AS "Phone",
    u.city AS "City",
    o.order_id AS "OrderID",
    o.paid_timestamp * 1000 AS "PurchaseTimestamp",
    o.amount / 100.0 AS "OrderAmount"
FROM product_data.orders o
JOIN product_data.users u ON u.user_id = o.user_id
WHERE o.product_id = 'PROD001';

Load the transformed data using Pandas:

import pandas as pd
from sqlalchemy import create_engine

def load_query_results(query):
    connection_string = 'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8mb4'.format(
        "db_username", "db_password", "127.0.0.1", "3306", "product_data"
    )
    engine = create_engine(connection_string)
    
    df = pd.read_sql(query, engine)
    df = df.astype({
        "UserID": int,
        "Nickname": str,
        "Gender": str,
        "Phone": str,
        "City": str,
        "OrderID": int,
        "PurchaseTimestamp": 'int64',
        "OrderAmount": float
    }, errors='ignore')
    
    return df

sql = '''
SELECT 
    u.user_id AS "UserID",
    u.nickname AS "Nickname",
    CASE u.gender 
        WHEN 0 THEN 'Female' 
        WHEN 1 THEN 'Male' 
        ELSE 'Unknown' 
    END AS "Gender",
    u.mobile AS "Phone",
    u.city AS "City",
    o.order_id AS "OrderID",
    o.paid_timestamp * 1000 AS "PurchaseTimestamp",
    o.amount / 100.0 AS "OrderAmount"
FROM product_data.orders o
JOIN product_data.users u ON u.user_id = o.user_id
WHERE o.product_id = 'PROD001'
'''

df = load_query_results(sql)
print(df)

Note: When running as a standalone script, timestamp fields may encounter issues with negative values when using int type. Using 'int64' resolves this issue.

Pandas-Based Transformation Alternative

If transformation occurs in Pandas instead of SQL:

import pandas as pd
from sqlalchemy import create_engine

def load_raw_data(query):
    connection_string = 'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8mb4'.format(
        "db_username", "db_password", "127.0.0.1", "3306", "product_data"
    )
    engine = create_engine(connection_string)
    
    df = pd.read_sql(query, engine)
    df = df.astype({
        "user_id": int,
        "nickname": str,
        "gender": str,
        "mobile": str,
        "city": str,
        "order_id": int,
        "paid_time": 'datetime64[ns]',
        "amount": float
    })
    
    df.columns = ["UserID", "Nickname", "Gender", "Phone", "City", "OrderID", "PurchaseTimestamp", "OrderAmount"]
    return df

sql = '''
SELECT 
    u.user_id,
    u.nickname,
    CASE u.gender 
        WHEN 0 THEN 'Female' 
        WHEN 1 THEN 'Male' 
        ELSE 'Unknown' 
    END AS gender,
    u.mobile,
    u.city,
    o.order_id,
    FROM_UNIXTIME(o.paid_timestamp) AS paid_time,
    o.amount / 100.0 AS amount
FROM product_data.orders o
JOIN product_data.users u ON u.user_id = o.user_id
WHERE o.product_id = 'PROD001'
'''

df = load_raw_data(sql)

Convert datetime columns to epoch milliseconds:

def convert_datetime_to_millis(df):
    datetime_columns = df.select_dtypes(include=['datetime64[ns]']).columns
    
    for col in datetime_columns:
        df[col] = (df[col] - pd.Timestamp("1970-01-01") - pd.Timedelta(hours=8)) // pd.Timedelta(milliseconds=1)
    
    return df

df = convert_datetime_to_millis(df)
print(df)

Converting to API Request Format

After field names and values are properly formatted, transform the data structure to match the Feishu batch insert API requirements. The API expects nested "fields" and "records" objects.

The API request structure:

{
    "records": [
        {
            "fields": {
                "FieldName": "value",
                "NumericField": 100,
                "SelectField": "Option1",
                "DateField": 1674206443000
            }
        }
    ]
}

Convert DataFrame to dictionary format:

def transform_to_request_payload(dataframe):
    records_dict = dataframe.to_dict(orient='records')
    wrapped_fields = [{'fields': record} for record in records_dict]
    payload = {'records': wrapped_fields}
    return payload

payload = transform_to_request_payload(df)
print(payload)

Inserting Data into Bitable

Insertion requires the app_token, table_id, and prepared request body:

import requests
import json

def insert_batch_records(app_token, table_id, payload, access_token):
    endpoint = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create"
    
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {access_token}'
    }
    
    request_body = json.dumps(payload).replace(': NaN', ': null')
    
    response = requests.post(endpoint, headers=headers, data=request_body)
    result = response.json()
    
    if result.get('code') == 0:
        record_count = len(payload['records'])
        print(f"Successfully inserted {record_count} records")
    else:
        raise RuntimeError(f"Insertion failed: {result.get('msg')}")


def main():
    app_token = 'APP_TOKEN_VALUE'
    table_id = 'TABLE_ID_VALUE'
    
    # df represents the processed data from earlier steps
    payload = transform_to_request_payload(df)
    insert_batch_records(app_token, table_id, payload, access_token)


if __name__ == '__main__':
    main()

Handling Edge Cases

Null Value Processing

When dataframe contains null values, special handling is required:

import pandas as pd
import json

test_data = pd.DataFrame({
    'text_field': [None, 'value_b', 'value_c'],
    'numeric_field': [1, 2, None]
})

serialized = json.dumps(test_data.to_dict(orient='records'))
print(serialized)

Text fields convert null to the string "null" (acceptable), while numeric fields produce "NaN" (unacceptable to Feishu API). Replace NaN with null:

sanitized_payload = serialized.replace(': NaN', ': null')

Batch Size Limits

The Feishu batch create API accepts maximum 500 records per request. For larger datasets, implement pagination:

def paginate_payload(dataframe, batch_size=500):
    records_dict = dataframe.to_dict(orient='records')
    
    payload_batches = []
    for start_idx in range(0, len(records_dict), batch_size):
        batch = records_dict[start_idx:start_idx + batch_size]
        wrapped = [{'fields': record} for record in batch]
        payload_batches.append({'records': wrapped})
    
    print(f"Data divided into {len(payload_batches)} batches")
    return payload_batches

Complete Integration Code

import requests
import json
import pandas as pd
from sqlalchemy import create_engine


def obtain_tenant_token(app_id, app_secret):
    auth_endpoint = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    
    payload = {"app_id": app_id, "app_secret": app_secret}
    response = requests.post(auth_endpoint, json=payload)
    token = response.json().get('tenant_access_token')
    print(f"Token obtained: {token[:10]}...")
    return token


def build_bitable(app_token, config, access_token):
    endpoint = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables"
    
    response = requests.post(endpoint, json=config, headers={
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {access_token}'
    })
    result = response.json()
    
    if result.get('code') == 0:
        table_id = result['data']['table_id']
        print(f"Table created: {config['table']['name']} ({table_id})")
        return table_id
    else:
        raise RuntimeError(f"Failed: {result.get('msg')}")


def fetch_data(query, connection_config, schema):
    engine = create_engine(connection_config)
    df = pd.read_sql(query, engine)
    df = df.astype(schema, errors='ignore')
    return df


def prepare_payload(dataframe, batch_limit=500):
    records_dict = dataframe.to_dict(orient='records')
    
    batches = []
    for offset in range(0, len(records_dict), batch_limit):
        segment = records_dict[offset:offset + batch_limit]
        wrapped = [{'fields': entry} for entry in segment]
        batches.append({'records': wrapped})
    
    print(f"Payload prepared: {len(batches)} batch(es)")
    return batches


def push_records(app_token, table_id, payload, access_token):
    endpoint = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/batch_create"
    
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {access_token}'
    }
    
    body = json.dumps(payload).replace(': NaN', ': null')
    
    response = requests.post(endpoint, headers=headers, data=body)
    result = response.json()
    
    if result.get('code') == 0:
        print(f"Inserted {len(payload['records'])} records successfully")
    else:
        print(result)
        raise RuntimeError(f"Insertion error: {result.get('msg')}")


def main():
    # Configuration
    app_id = 'YOUR_APP_ID'
    app_secret = 'YOUR_APP_SECRET'
    app_token = 'YOUR_APP_TOKEN'
    
    access_token = obtain_tenant_token(app_id, app_secret)
    
    # Create table
    table_config = {
        "table": {
            "name": "Product Customer List",
            "default_view_name": "All Records View",
            "fields": [
                {"field_name": "UserID", "type": 2, "ui_type": "Number", "property": {"formatter": "0"}},
                {"field_name": "Nickname", "type": 1, "ui_type": "Text"},
                {"field_name": "Gender", "type": 3, "ui_type": "SingleSelect"},
                {"field_name": "Phone", "type": 13, "ui_type": "Phone"},
                {"field_name": "City", "type": 3, "ui_type": "SingleSelect"},
                {"field_name": "OrderID", "type": 1, "ui_type": "Text"},
                {"field_name": "PurchaseTimestamp", "type": 5, "ui_type": "DateTime", "property": {"date_formatter": "yyyy/MM/dd HH:mm", "auto_fill": False}},
                {"field_name": "OrderAmount", "type": 2, "ui_type": "Number", "property": {"formatter": "0.00"}}
            ]
        }
    }
    
    table_id = build_bitable(app_token, table_config, access_token)
    
    # Query data
    query = '''
        SELECT 
            u.user_id AS "UserID",
            u.nickname AS "Nickname",
            CASE u.gender 
                WHEN 0 THEN 'Female' 
                WHEN 1 THEN 'Male' 
                ELSE 'Unknown' 
            END AS "Gender",
            u.mobile AS "Phone",
            u.city AS "City",
            o.order_id AS "OrderID",
            o.paid_timestamp * 1000 AS "PurchaseTimestamp",
            o.amount / 100.0 AS "OrderAmount"
        FROM product_data.orders o
        JOIN product_data.users u ON u.user_id = o.user_id
        WHERE o.product_id = 'PROD001'
    '''
    
    connection_string = 'mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8mb4'.format(
        "DB_USER", "DB_PASSWORD", "127.0.0.1", "3306", "product_data"
    )
    
    field_schema = {
        "UserID": int,
        "Nickname": str,
        "Gender": str,
        "Phone": str,
        "City": str,
        "OrderID": int,
        "PurchaseTimestamp": int,
        "OrderAmount": float
    }
    
    df = fetch_data(query, connection_string, field_schema)
    
    # Prepare and insert
    payloads = prepare_payload(df, batch_limit=500)
    
    for batch_payload in payloads:
        push_records(app_token, table_id, batch_payload, access_token)


if __name__ == '__main__':
    main()

Configure with valid Feishu app credentials and database connection details before execution.

Tags: Feishu API python MySQL data pipeline Bitable

Posted on Tue, 29 Sep 2026 16:33:01 +0000 by sanchez77

Hot Tags

©2026 Freaks City