Django框架连接数据库

这里以同时连接sqlite、mysql、redis为例
首先要下载对应的sqlite、mysql、redis相应的模块插件
开始上代码
settings.py 配置文件

import os
# 引入mysql模块
import pymysql
# pymysql.install_as_MySQLdb()来确保pymysql能够兼容这些代码
pymysql.install_as_MySQLdb()# 数据库连接参数配置DATABASES = {'default': {'ENGINE': 'django.db.backends.sqlite3','NAME': BASE_DIR / 'db.sqlite3', # 本地sqlite数据文件地址},#mysql 测试服务器参数配置'mysql': {'ENGINE': 'django.db.backends.mysql','NAME': 'root','USER': 'root','PASSWORD': 'root','HOST': '127.0.0.1','PORT': '3306',}
}
# 配置测试redis参数 DATABASES不支持redis 只能单独配置
COUSTOM_REDIS_CONFIG = {'PASSWORD': 'root','HOST': '127.0.0.1','PORT':6379,
}#【弃用】**** 这里的参数没有研究明白 刚感兴趣或者明白的小伙伴可以评论交流沟通下 一块成长 **** 
#【弃用】**** 这里的参数没有研究明白 刚感兴趣或者明白的小伙伴可以评论交流沟通下 一块成长 **** 
#【弃用】**** 这里的参数没有研究明白 刚感兴趣或者明白的小伙伴可以评论交流沟通下 一块成长 **** # 配置Redis作为缓存后端 【无法使用 key是 ascribe_to:pay_report_id 自动变成 1:ascribe_to:pay_report_id】
# from django.core.cache import caches  # 逻辑文件中 导入caches对象
# r = caches['alt_db3'] # 逻辑文件中 使用备用的Redis连接（第3个数据库）
CACHES = {'default': {'BACKEND': 'django_redis.cache.RedisCache','LOCATION': 'redis://180.184.93.11:6379/0',#配置指向第1个数据库（索引为0）'KEY_PREFIX': '',  # 默认键的前缀'OPTIONS': {'CLIENT_CLASS': 'django_redis.client.DefaultClient','PASSWORD': 'xuannao!@#456',}},'alt_db3': {'BACKEND': 'django_redis.cache.RedisCache','LOCATION': 'redis://180.184.93.11:6379/2',  # 配置指向Redis的第3个数据库（索引为2)'KEY_PREFIX': '090',  # 默认键的前缀'OPTIONS': {'CLIENT_CLASS': 'django_redis.client.DefaultClient','PASSWORD': 'xuannao!@#456',}}
}

test.py 逻辑文件

# 导入对应的模块
from django.conf import settings
import redis
from django.db import connections
import sqlite3
from advert.models import *
from django.db.models import CharField, IntegerField, Value
from django.db.models.functions import Length
from django.http import JsonResponse
import time# 连接mysql
mysql_connection = connections['mysql']# 或者 连接mysql
# mysql_connection = MySQLdb.connect(host="localhost", user="yourusername", passwd="yourpassword", db="yourdatabase")# 连接redis
r_host = settings.COUSTOM_REDIS_CONFIG['HOST']
r_port = settings.COUSTOM_REDIS_CONFIG['PORT']
r_pwd = settings.COUSTOM_REDIS_CONFIG['PASSWORD']
r = redis.Redis(r_host, r_port, 2, r_pwd)# 连接sqlite
sqlite_connection = sqlite3.connect(settings.DATABASES['default']['NAME'])
sqlite_cursor = sqlite_connection.cursor()# 获取mysql数据 以原生sql方式举例
# 为了防止数据库查询发生 SQL 注入的攻击，我们可以使用 %s 占位符来转义查询的条件：
sql = "select id,user_id from ln_user_recharge where status = 2 and id > %s order by id asc"
cursor.execute(sql, (pay_order_id,)) # pay_order_id 作为id的条件
rows = cursor.fetchall() # 查询全部结果 更多查询方式 参考python或者Django手册即可# 获取sqlite数据 以原生sql方式举例 和mysql类似
# 执行查询 获取sqlite中用户的oaid 和 渠道
sqlite_sql = "SELECT oaid,uid,channel FROM advert_clientreport WHERE uid = ? and LENGTH(oaid) > 30"
sqlite_cursor.execute(sqlite_sql, (user_id,))
# 获取查询结果
sqlite_results = sqlite_cursor.fetchall()# 获取sqlite数据 以models的方式 基于Django框架 更多请参考Django框架
reporting = ClientReport.objects.annotate(oaid_length=Length('oaid', output_field=IntegerField())).filter(uid=user_id,oaid_length__gt=30)# redis举例
r.set('key','value')
r.get('key')

models.py 数据库文件

from django.db import models
from django.db.models import Sum# sqlite 数据表3
class ClientReport(models.Model):oaid = models.CharField(max_length=128, verbose_name="OAID设备号", null=True, blank=True)idfa = models.CharField(max_length=128, verbose_name="IDFA识别码", null=True, blank=True)imei = models.CharField(max_length=128, verbose_name="IMEI序列号", null=True, blank=True)mac = models.CharField(max_length=128, verbose_name="MAC地址", null=True, blank=True)uip = models.CharField(max_length=64, verbose_name="IP地址", null=True, blank=True)ua = models.CharField(max_length=255, verbose_name="UserAgent", null=True, blank=True)uid = models.IntegerField(verbose_name='新用户ID', null=True, blank=True, db_index=True)version = models.CharField(max_length=32, verbose_name="版本号", null=True, blank=True)client = models.CharField(max_length=32, verbose_name="客户端", default='1', choices=CLIENT_DICT)channel = models.CharField(max_length=32, verbose_name="下载渠道", null=True, blank=True)sex = models.CharField(max_length=8, verbose_name="性别", null=True, blank=True, choices=SEX_DICT)is_true = models.BooleanField(default=True, verbose_name="是否真实")isvalid = models.BooleanField(default=False, verbose_name="是否有效")create_time = models.DateTimeField(verbose_name='创建时间', auto_now_add=True, null=True)class Meta:app_label = 'advert'verbose_name_plural = '客户端上报'