dataset，一个非常好用的 Python 库！

大家好，今天为大家分享一个非常好用的 Python 库 - dataset。

Github地址：https://github.com/pudo/dataset

dataset是一个专为Python开发者设计的轻量级数据库操作库，它将复杂的SQL语句转化为简洁直观的Python代码。该库的核心理念是"数据库操作应该像操作Python字典一样简单"，通过提供类似NoSQL的操作接口来简化关系型数据库的使用。dataset支持SQLite、MySQL、PostgreSQL等主流数据库，无需编写繁琐的SQL语句或复杂的ORM配置，就能实现数据的增删改查操作。特别适合数据分析、快速原型开发、小型应用以及需要频繁进行数据库操作的场景，让开发者能够专注于业务逻辑而非数据库语法细节。

安装

1、安装方法

# 安装基础版本
pip install dataset
​
# 安装支持MySQL的版本
pip install dataset[mysql]
​
# 安装支持PostgreSQL的版本  
pip install dataset[postgresql]
​
# 安装完整版本（支持所有数据库）
pip install dataset[complete]

2、验证安装

import dataset
print(f"dataset版本: {dataset.__version__}")
​
# 创建内存数据库测试
db = dataset.connect('sqlite:///:memory:')
print(f"数据库连接成功: {type(db)}")

核心特性

• 简洁的API设计：提供类似字典操作的直观接口

• 多数据库支持：兼容SQLite、MySQL、PostgreSQL等主流数据库

• 自动表创建：根据数据结构自动创建数据表和字段

• 灵活的查询方式：支持多种数据检索和过滤方法

• 事务处理：内置事务管理确保数据一致性

• 数据导出功能：支持将查询结果导出为多种格式

• 无需预定义模式：动态适应数据结构变化

• 轻量级设计：最小化依赖，快速部署

基本功能

1、数据库连接与表操作

dataset提供了统一的连接接口，支持多种数据库类型，通过简单的URL格式即可建立连接，无需复杂的配置过程，连接建立后，可以直接通过表名访问数据表，如果表不存在会自动创建。

import dataset
​
# 连接SQLite数据库
db = dataset.connect('sqlite:///example.db')
​
# 连接MySQL数据库（需要安装mysql-python或pymysql）
# db = dataset.connect('mysql://user:password@localhost/database')
​
# 获取或创建表
users_table = db['users']
print(f"表对象类型: {type(users_table)}")
​
# 查看数据库中的所有表
table_names = db.tables
print(f"数据库中的表: {list(table_names)}")
​
# 获取表结构信息
columns = users_table.columns
print(f"users表的字段: {list(columns)}")

2、数据插入与更新

dataset允许通过字典格式直接插入数据，自动处理字段类型推断和表结构更新。支持单条记录插入和批量插入操作，同时提供了upsert功能，能够智能判断是插入新记录还是更新已有记录。

# 插入单条记录
users_table.insert({
    'name': '张三',
    'email': 'zhangsan@example.com',
    'age': 25,
    'city': '北京'
})
​
# 批量插入数据
user_data = [
    {'name': '李四', 'email': 'lisi@example.com', 'age': 30, 'city': '上海'},
    {'name': '王五', 'email': 'wangwu@example.com', 'age': 28, 'city': '广州'},
    {'name': '赵六', 'email': 'zhaoliu@example.com', 'age': 35, 'city': '深圳'}
]
users_table.insert_many(user_data)
​
# 使用upsert（插入或更新）
users_table.upsert({
    'name': '张三',
    'email': 'zhangsan@gmail.com',  # 更新邮箱
    'age': 26,  # 更新年龄
    'city': '北京'
}, ['name'])  # 以name为唯一标识
​
print("数据插入完成")
print(f"表中记录数: {len(users_table)}")

3、数据查询与筛选

dataset提供了多种灵活的查询方式，从简单的全表查询到复杂的条件筛选。支持链式操作、排序、限制返回数量等功能，查询结果以字典格式返回，便于后续处理和分析。

# 查询所有数据
all_users = list(users_table.all())
print(f"所有用户: {len(all_users)} 条记录")
​
# 条件查询
beijing_users = users_table.find(city='北京')
for user in beijing_users:
    print(f"北京用户: {user['name']} - {user['email']}")
​
# 复杂条件查询
young_users = users_table.find(users_table.table.columns.age < 30)
print(f"30岁以下用户数量: {len(list(young_users))}")
​
# 查询单条记录
user = users_table.find_one(name='张三')
if user:
    print(f"找到用户: {user['name']}, 年龄: {user['age']}")
​
# 使用SQL查询（更复杂的查询）
results = db.query('SELECT city, COUNT(*) as count FROM users GROUP BY city')
for row in results:
    print(f"城市: {row['city']}, 人数: {row['count']}")

高级功能

1、事务处理与数据完整性

dataset提供了简洁的事务管理接口，通过上下文管理器自动处理事务的开始、提交和回滚操作。

# 使用事务处理批量操作
try:
    with db.transaction():
        # 创建订单表
        orders_table = db['orders']
        
        # 插入订单数据
        order_data = {
            'user_id': 1,
            'product_name': 'Python编程书籍',
            'quantity': 2,
            'price': 89.90,
            'status': 'pending'
        }
        order_id = orders_table.insert(order_data)
        
        # 更新用户信息
        users_table.update({'id': 1, 'last_order_time': '2025-01-15'}, ['id'])
        
        # 如果发生异常，事务会自动回滚
        print(f"订单创建成功，ID: {order_id}")
        
except Exception as e:
    print(f"事务执行失败: {e}")
​
# 检查事务结果
order_count = len(db['orders'])
print(f"订单表中记录数: {order_count}")

2、数据导出与格式转换

数据导出功能让dataset成为优秀的数据处理工具，支持将查询结果导出为多种格式，包括CSV、JSON、Excel等，便于数据分析和报告生成。

import json
​
# 导出数据为JSON格式
all_users = list(users_table.all())
with open('users_export.json', 'w', encoding='utf-8') as f:
    json.dump(all_users, f, ensure_ascii=False, indent=2)
​
# 导出特定字段
user_summary = []
for user in users_table.find():
    user_summary.append({
        'name': user['name'],
        'city': user['city'],
        'age': user['age']
    })
​
# 导出为CSV（需要安装pandas）
try:
    import pandas as pd
    df = pd.DataFrame(user_summary)
    df.to_csv('users_summary.csv', index=False, encoding='utf-8')
    print("CSV导出成功")
except ImportError:
    print("需要安装pandas才能导出CSV格式")
​
# 统计查询并导出
city_stats = db.query('''
    SELECT city, 
           COUNT(*) as user_count, 
           AVG(age) as avg_age 
    FROM users 
    GROUP BY city
''')
​
stats_data = []
for row in city_stats:
    stats_data.append({
        'city': row['city'],
        'user_count': row['user_count'],
        'average_age': round(row['avg_age'], 2)
    })
​
print("城市统计数据:", stats_data)

实际应用场景

1、数据采集与存储系统

在数据采集项目中，经常需要将从不同源头获取的数据统一存储到数据库中，dataset的灵活性使其成为理想的选择，能够自动适应不同的数据结构，无需预先定义复杂的数据模型。

以下示例展示了一个完整的网络数据采集与存储流程：

import requests
import time
from datetime import datetime
​
def collect_and_store_data():
    """数据采集与存储示例"""
    # 连接数据库
    db = dataset.connect('sqlite:///data_collection.db')
    
    # 创建数据表
    api_data_table = db['api_responses']
    log_table = db['collection_logs']
    
    # 模拟API数据采集
    api_urls = [
        'https://httpbin.org/json',
        'https://httpbin.org/uuid',
        'https://httpbin.org/ip'
    ]
    
    collected_count = 0
    
    for url in api_urls:
        try:
            # 记录采集开始
            log_entry = {
                'url': url,
                'start_time': datetime.now().isoformat(),
                'status': 'started'
            }
            log_id = log_table.insert(log_entry)
            
            # 请求API数据
            response = requests.get(url, timeout=10)
            response.raise_for_status()
            
            # 存储响应数据
            data_entry = {
                'url': url,
                'response_data': response.text,
                'status_code': response.status_code,
                'collected_at': datetime.now().isoformat(),
                'data_size': len(response.text)
            }
            api_data_table.insert(data_entry)
            
            # 更新日志状态
            log_table.update({
                'id': log_id,
                'end_time': datetime.now().isoformat(),
                'status': 'success'
            }, ['id'])
            
            collected_count += 1
            print(f"✓ 成功采集: {url}")
            
            # 避免请求过快
            time.sleep(1)
            
        except Exception as e:
            # 记录错误
            log_table.update({
                'id': log_id,
                'end_time': datetime.now().isoformat(),
                'status': 'error',
                'error_message': str(e)
            }, ['id'])
            print(f"✗ 采集失败: {url} - {e}")
    
    print(f"\n采集完成，成功 {collected_count}/{len(api_urls)} 个API")
    
    # 查看采集结果统计
    stats = db.query('''
        SELECT status, COUNT(*) as count 
        FROM collection_logs 
        GROUP BY status
    ''')
    
    for stat in stats:
        print(f"状态 {stat['status']}: {stat['count']} 次")
​
# 执行数据采集
collect_and_store_data()

2、配置管理与用户偏好系统

dataset提供了简单高效的解决方案，能够动态存储各种配置信息，支持版本控制和历史记录。

以下示例演示了一个完整的配置管理系统实现：

import json
from datetime import datetime
​
class ConfigManager:
    """配置管理器"""
    
    def __init__(self, db_path='config.db'):
        self.db = dataset.connect(f'sqlite:///{db_path}')
        self.configs_table = self.db['user_configs']
        self.history_table = self.db['config_history']
    
    def set_config(self, user_id, category, key, value):
        """设置配置项"""
        config_data = {
            'user_id': user_id,
            'category': category,
            'config_key': key,
            'config_value': json.dumps(value),
            'updated_at': datetime.now().isoformat()
        }
        
        # 保存历史记录
        existing = self.configs_table.find_one(
            user_id=user_id, 
            category=category, 
            config_key=key
        )
        
        if existing:
            self.history_table.insert({
                'user_id': user_id,
                'category': category,
                'config_key': key,
                'old_value': existing['config_value'],
                'new_value': config_data['config_value'],
                'changed_at': datetime.now().isoformat()
            })
        
        # 更新配置
        self.configs_table.upsert(config_data, ['user_id', 'category', 'config_key'])
        print(f"配置已更新: {user_id}/{category}/{key}")
    
    def get_config(self, user_id, category, key, default=None):
        """获取配置项"""
        config = self.configs_table.find_one(
            user_id=user_id,
            category=category,
            config_key=key
        )
        
        if config:
            return json.loads(config['config_value'])
        return default
    
    def get_user_configs(self, user_id):
        """获取用户所有配置"""
        configs = {}
        for config in self.configs_table.find(user_id=user_id):
            category = config['category']
            key = config['config_key']
            value = json.loads(config['config_value'])
            
            if category not in configs:
                configs[category] = {}
            configs[category][key] = value
        
        return configs
​
# 使用配置管理器
config_mgr = ConfigManager()
​
# 设置用户配置
config_mgr.set_config('user_001', 'ui', 'theme', 'dark')
config_mgr.set_config('user_001', 'ui', 'language', 'zh-CN')
config_mgr.set_config('user_001', 'notification', 'email_alerts', True)
config_mgr.set_config('user_001', 'notification', 'sound_enabled', False)
​
# 获取配置
theme = config_mgr.get_config('user_001', 'ui', 'theme')
print(f"用户主题设置: {theme}")
​
# 获取所有配置
all_configs = config_mgr.get_user_configs('user_001')
print(f"用户所有配置: {json.dumps(all_configs, indent=2, ensure_ascii=False)}")

总结

dataset作为Python数据库操作的优秀工具，以其简洁直观的API设计和强大的功能特性，为开发者提供了高效的数据存储解决方案。通过本文的全面介绍，我们了解了dataset从基础操作到高级应用的完整使用方法。它成功地将复杂的SQL操作转化为简单的Python字典操作，大大降低了数据库编程的门槛。dataset的自动表创建、灵活的查询接口、完善的事务处理和多格式数据导出功能，使其在快速原型开发、数据采集、配置管理等场景中表现出色。相比传统的ORM框架，dataset更加轻量级且易于上手，特别适合中小型项目和数据分析任务。在实际开发中，合理运用dataset可以显著提升数据操作的效率和代码的可维护性，是Python开发者工具箱中的重要利器。

AI工具的成熟，让程序员也有了以前不敢想象的能力。海外市场的广阔，给了我们更大的舞台。

如果你也在考虑新的出路，如果你也想尝试AI编程出海这个方向，欢迎加入我们。

扫码或搜索 257735 添加微信，发送暗号「美金」，了解详细信息。