当前位置：首页 > news >正文

数据库完整指南：从基础到 Django 集成

news 2025/10/7 6:44:33

数据库基础概念

什么是数据库？

数据库是结构化信息的集合，用于高效存储、管理和检索数据。

数据库类型

关系型数据库：MySQL、PostgreSQL、SQLite
非关系型数据库：MongoDB、Redis
时序数据库：InfluxDB
图数据库：Neo4j

ACID 原则

原子性：事务要么全部完成，要么全部不完成
一致性：事务必须使数据库从一种一致状态变为另一种一致状态
隔离性：并发事务之间互不干扰
持久性：事务完成后，对数据库的修改是永久的

关系型数据库

核心概念

sql

-- 数据库操作
CREATE DATABASE mydb;
USE mydb;-- 表操作
CREATE TABLE users (id INT PRIMARY KEY AUTO_INCREMENT,username VARCHAR(50) UNIQUE NOT NULL,email VARCHAR(100) NOT NULL,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);-- 数据操作
INSERT INTO users (username, email) VALUES ('john', 'john@example.com');
SELECT * FROM users WHERE username = 'john';
UPDATE users SET email = 'new@example.com' WHERE id = 1;
DELETE FROM users WHERE id = 1;

常见关系型数据库比较

数据库	特点	适用场景
MySQL	流行、成熟、社区强大	Web应用、中小型系统
PostgreSQL	功能丰富、标准兼容性好	复杂应用、地理数据
SQLite	轻量级、零配置	移动应用、小型项目
Oracle	企业级、功能强大	大型企业系统

Django 数据库配置

基本配置

python

# settings.py
DATABASES = {'default': {'ENGINE': 'django.db.backends.sqlite3','NAME': BASE_DIR / 'db.sqlite3',}
}

PostgreSQL 配置

python

# settings.py
DATABASES = {'default': {'ENGINE': 'django.db.backends.postgresql','NAME': 'mydatabase','USER': 'mydatabaseuser','PASSWORD': 'mypassword','HOST': 'localhost','PORT': '5432','CONN_MAX_AGE': 600,  # 连接存活时间（秒）'OPTIONS': {'connect_timeout': 10,}}
}

MySQL 配置

python

# settings.py
DATABASES = {'default': {'ENGINE': 'django.db.backends.mysql','NAME': 'mydatabase','USER': 'mydatabaseuser','PASSWORD': 'mypassword','HOST': 'localhost','PORT': '3306','OPTIONS': {'charset': 'utf8mb4','init_command': "SET sql_mode='STRICT_TRANS_TABLES'",},'TEST': {'CHARSET': 'utf8mb4','COLLATION': 'utf8mb4_unicode_ci',}}
}

多数据库配置

python

# settings.py
DATABASES = {'default': {'ENGINE': 'django.db.backends.postgresql','NAME': 'primary_db',# ... 其他配置},'read_replica': {'ENGINE': 'django.db.backends.postgresql','NAME': 'replica_db',# ... 其他配置},'logging': {'ENGINE': 'django.db.backends.sqlite3','NAME': BASE_DIR / 'logs.db',}
}# 数据库路由
DATABASE_ROUTERS = ['myapp.routers.DatabaseRouter']

Django 模型详解

基础模型定义

python

# models.py
from django.db import models
from django.urls import reverse
from django.core.validators import MinLengthValidator, MaxValueValidatorclass TimestampModel(models.Model):"""抽象基类，提供时间戳字段"""created_at = models.DateTimeField(auto_now_add=True)updated_at = models.DateTimeField(auto_now=True)class Meta:abstract = Trueclass User(TimestampModel):# 字段类型username = models.CharField(max_length=50,unique=True,validators=[MinLengthValidator(3)],verbose_name='用户名')email = models.EmailField(unique=True, verbose_name='邮箱')age = models.PositiveIntegerField(null=True,blank=True,validators=[MaxValueValidator(150)],verbose_name='年龄')is_active = models.BooleanField(default=True, verbose_name='是否激活')birth_date = models.DateField(null=True, blank=True, verbose_name='出生日期')salary = models.DecimalField(max_digits=10,decimal_places=2,null=True,blank=True,verbose_name='薪资')bio = models.TextField(blank=True, verbose_name='个人简介')avatar = models.ImageField(upload_to='avatars/',null=True,blank=True,verbose_name='头像')# 元数据class Meta:verbose_name = '用户'verbose_name_plural = '用户'ordering = ['-created_at']indexes = [models.Index(fields=['username']),models.Index(fields=['email', 'is_active']),]constraints = [models.CheckConstraint(check=models.Q(age__gte=0),name='age_positive')]def __str__(self):return self.usernamedef get_absolute_url(self):return reverse('user_detail', kwargs={'pk': self.pk})@propertydef display_name(self):return f"{self.username} ({self.email})"def save(self, *args, **kwargs):# 保存前的自定义逻辑self.email = self.email.lower()super().save(*args, **kwargs)

字段类型详解

字段类型	说明	对应数据库类型
CharField	字符串字段	VARCHAR
TextField	大文本字段	TEXT
IntegerField	整数字段	INTEGER
DecimalField	十进制数字段	NUMERIC
BooleanField	布尔字段	BOOLEAN
DateTimeField	日期时间字段	DATETIME
DateField	日期字段	DATE
TimeField	时间字段	TIME
EmailField	邮箱字段	VARCHAR
URLField	URL字段	VARCHAR
FileField	文件字段	VARCHAR
ImageField	图片字段	VARCHAR

字段选项

python

class ExampleModel(models.Model):# 常用字段选项name = models.CharField(max_length=100,null=True,           # 数据库允许NULLblank=True,          # 表单验证允许空值default='Unknown',   # 默认值unique=True,         # 唯一值db_index=True,       # 创建索引choices=[            # 选择项('draft', '草稿'),('published', '已发布'),],help_text='请输入名称',  # 帮助文本verbose_name='名称'     # 可读名称)

数据库迁移

迁移命令

bash

# 创建迁移文件
python manage.py makemigrations# 查看迁移SQL
python manage.py sqlmigrate myapp 0001# 应用迁移
python manage.py migrate# 查看迁移状态
python manage.py showmigrations# 撤销迁移
python manage.py migrate myapp 0001# 数据迁移
python manage.py makemigrations --empty myapp

自定义迁移

python

# myapp/migrations/0002_custom_migration.py
from django.db import migrations, modelsdef add_default_categories(apps, schema_editor):Category = apps.get_model('myapp', 'Category')Category.objects.bulk_create([Category(name='技术'),Category(name='生活'),Category(name='旅游'),])def remove_default_categories(apps, schema_editor):Category = apps.get_model('myapp', 'Category')Category.objects.filter(name__in=['技术', '生活', '旅游']).delete()class Migration(migrations.Migration):dependencies = [('myapp', '0001_initial'),]operations = [migrations.RunPython(add_default_categories,remove_default_categories),]

数据库查询

基础查询

python

from myapp.models import User, Post
from django.db.models import Q, Count, Avg, Sum# 创建
user = User.objects.create(username='john', email='john@example.com')# 查询所有
users = User.objects.all()# 过滤查询
active_users = User.objects.filter(is_active=True)
john_users = User.objects.filter(username__contains='john')
recent_users = User.objects.filter(created_at__gte='2024-01-01')# 排除查询
inactive_users = User.objects.exclude(is_active=True)# 获取单个对象
try:user = User.objects.get(id=1)
except User.DoesNotExist:user = None# 快捷方法
user = User.objects.filter(id=1).first()# 复杂查询
complex_query = User.objects.filter(Q(is_active=True) & (Q(username__startswith='j') | Q(email__contains='example'))
)# 排序
users_ordered = User.objects.order_by('-created_at', 'username')# 切片
recent_10_users = User.objects.order_by('-created_at')[:10]

高级查询方法

python

# 聚合查询
from django.db.models import Count, Avg, Max, Minstats = User.objects.aggregate(total_users=Count('id'),avg_age=Avg('age'),max_salary=Max('salary')
)# 分组查询
user_posts_count = User.objects.annotate(post_count=Count('posts')
).filter(post_count__gt=0)# 值列表
usernames = User.objects.values_list('username', flat=True)
user_data = User.objects.values('username', 'email')# 选择相关对象（减少查询）
users_with_posts = User.objects.select_related('profile').prefetch_related('posts')# 原生SQL查询
from django.db import connectionwith connection.cursor() as cursor:cursor.execute("SELECT * FROM myapp_user WHERE age > %s", [18])rows = cursor.fetchall()# 使用raw方法
users = User.objects.raw('SELECT * FROM myapp_user WHERE is_active = %s', [True])

查询性能优化

python

# 不好的查询 - N+1 问题
users = User.objects.all()
for user in users:print(user.profile.bio)  # 每次循环都会查询数据库# 好的查询 - 使用select_related
users = User.objects.select_related('profile').all()
for user in users:print(user.profile.bio)  # 一次性查询# 使用prefetch_related处理多对多关系
posts = Post.objects.prefetch_related('tags').all()
for post in posts:print(post.tags.all())  # 一次性预加载

数据库关系

一对一关系

python

class UserProfile(models.Model):user = models.OneToOneField(User,on_delete=models.CASCADE,related_name='profile')bio = models.TextField(blank=True)location = models.CharField(max_length=100, blank=True)def __str__(self):return f"{self.user.username}的档案"# 使用示例
user = User.objects.get(id=1)
profile = user.profile  # 通过related_name访问

一对多关系

python

class Post(TimestampModel):author = models.ForeignKey(User,on_delete=models.CASCADE,related_name='posts')title = models.CharField(max_length=200)content = models.TextField()is_published = models.BooleanField(default=False)class Meta:ordering = ['-created_at']def __str__(self):return self.title# 使用示例
user = User.objects.get(id=1)
user_posts = user.posts.all()  # 用户的所有文章
latest_post = user.posts.first()

多对多关系

python

class Tag(models.Model):name = models.CharField(max_length=50, unique=True)created_at = models.DateTimeField(auto_now_add=True)def __str__(self):return self.nameclass Post(TimestampModel):# ... 其他字段tags = models.ManyToManyField(Tag,related_name='posts',blank=True)def get_tag_names(self):return list(self.tags.values_list('name', flat=True))# 使用示例
post = Post.objects.get(id=1)
post.tags.add(tag1, tag2)  # 添加标签
post.tags.remove(tag1)     # 移除标签
post.tags.clear()          # 清空所有标签

自定义中间表

python

class PostTag(models.Model):post = models.ForeignKey(Post, on_delete=models.CASCADE)tag = models.ForeignKey(Tag, on_delete=models.CASCADE)added_by = models.ForeignKey(User, on_delete=models.CASCADE)added_at = models.DateTimeField(auto_now_add=True)class Meta:unique_together = ['post', 'tag']class Post(models.Model):tags = models.ManyToManyField(Tag,through='PostTag',related_name='posts')

数据库优化

索引优化

python

class User(models.Model):username = models.CharField(max_length=50, db_index=True)email = models.EmailField(unique=True)class Meta:indexes = [models.Index(fields=['username', 'email']),models.Index(fields=['-created_at']),]

查询优化技巧

python

# 1. 使用only和defer选择特定字段
users = User.objects.only('username', 'email')  # 只查询指定字段
users = User.objects.defer('bio')  # 排除指定字段# 2. 使用exists检查存在性
if User.objects.filter(username='john').exists():# 处理逻辑# 3. 使用count获取数量
user_count = User.objects.count()# 4. 使用bulk操作
users = [User(username=f'user{i}') for i in range(1000)]
User.objects.bulk_create(users)  # 批量创建# 5. 使用update批量更新
User.objects.filter(is_active=False).update(is_active=True)

数据库连接池

python

# 使用django-db-connection-pool
DATABASES = {'default': {'ENGINE': 'dj_db_conn_pool.backends.mysql','NAME': 'mydatabase','POOL_OPTIONS': {'POOL_SIZE': 10,'MAX_OVERFLOW': 20,'RECYCLE': 3600,}}
}

生产环境部署

数据库备份

python

# management/commands/backup_database.py
from django.core.management.base import BaseCommand
from django.conf import settings
import subprocess
import os
from datetime import datetimeclass Command(BaseCommand):help = '备份数据库'def handle(self, *args, **options):timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')backup_file = f'backup_{timestamp}.sql'if settings.DATABASES['default']['ENGINE'] == 'django.db.backends.postgresql':cmd = ['pg_dump','-h', settings.DATABASES['default']['HOST'],'-U', settings.DATABASES['default']['USER'],'-d', settings.DATABASES['default']['NAME'],'-f', backup_file]env = os.environ.copy()env['PGPASSWORD'] = settings.DATABASES['default']['PASSWORD']subprocess.run(cmd, env=env)self.stdout.write(self.style.SUCCESS(f'备份完成: {backup_file}'))

数据库监控

python

# 使用Django Debug Toolbar
INSTALLED_APPS = [# ...'debug_toolbar',
]MIDDLEWARE = ['debug_toolbar.middleware.DebugToolbarMiddleware',# ...
]# 数据库查询日志
LOGGING = {'version': 1,'handlers': {'console': {'level': 'DEBUG','class': 'logging.StreamHandler',},},'loggers': {'django.db.backends': {'level': 'DEBUG','handlers': ['console'],}}
}

生产环境配置

python

# settings/production.py
DATABASES = {'default': {'ENGINE': 'django.db.backends.postgresql','NAME': os.getenv('DB_NAME'),'USER': os.getenv('DB_USER'),'PASSWORD': os.getenv('DB_PASSWORD'),'HOST': os.getenv('DB_HOST'),'PORT': os.getenv('DB_PORT', '5432'),'CONN_MAX_AGE': 600,'OPTIONS': {'sslmode': 'require',}}
}# 数据库性能配置
DATABASES['default']['ATOMIC_REQUESTS'] = False  # 提高性能
DATABASES['default']['AUTOCOMMIT'] = True

数据库健康检查

python

# health_checks.py
from django.db import connection
from django.http import JsonResponsedef database_health_check(request):try:with connection.cursor() as cursor:cursor.execute("SELECT 1")result = cursor.fetchone()if result[0] == 1:return JsonResponse({'status': 'healthy'})else:return JsonResponse({'status': 'unhealthy'}, status=500)except Exception as e:return JsonResponse({'status': 'error', 'message': str(e)}, status=500)