非Django项目使用Django ORM方法
概述
平常项目中主要使用Django框架做API接口开发,对数据库的操作都是使用Django的ORM框架。对于非API服务就不适合使用Django进行开发,操作数据库就使用了SQLAlchemy框架进行处理。
这样就造成了同一个子系统中使用了两套ORM框架,对同一张表要使用不同的ORM框架定义两次,增加了学习和维护成本。因此考虑将Django框架的ORM脱离Django进行使用的可行性。
解决方案
例子如下:
# -*- coding: utf-8 -*-
from django.conf import settings
# 初始化Django的配置,引用DjangoORM之前必须要先进行此配置。所有配置都和Django中一样的定义。
settings.configure(
**{
'DEBUG': True,
'DATABASES': {
'default': {
'ENGINE': 'django.db.backends.mysql',
'NAME': 'cmp_portal',
'USER': 'root',
'PASSWORD': 'aaaaaa',
'HOST': '127.0.0.1',
'PORT': '3306'
},
},
}
)
from django.db import close_old_connections
from portal_test import models
import random
def update():
while True:
try:
# 进行数据库操作前要调用此函数,避免使用了失效的链接
close_old_connections()
obj = models.User.objects.first()
obj.name = obj.name + 1
obj.save()
print("update:", object)
# time.sleep(random.random()*0.5)
except Exception as err:
print("update:", "-" * 40)
print(err)
break
def create():
while True:
try:
# 进行数据库操作前要调用此函数,避免使用了失效的链接
close_old_connections()
obj = models.User.objects.create(name=random.randint(0, 100))
print("create:", obj)
# time.sleep(random.random() * 0.5)
except Exception as err:
print("create:", "-" * 40)
print(err)
def select():
while True:
try:
# 进行数据库操作前要调用此函数,避免使用了失效的链接
close_old_connections()
print("select:", models.User.objects.all()[:5])
# time.sleep(0.5)
except Exception as err:
print("select:", "-" * 40)
print(err)
break
def delete():
while True:
try:
# 进行数据库操作前要调用此函数,避免使用了失效的链接
close_old_connections()
obj = models.User.objects.first()
print("delete:", obj)
obj.delete()
# time.sleep(0.5)
except Exception as err:
print("delete:", "-" * 40)
print(err)
break
注意事项:
- 引用DjangoORM之前必须要先初始化Django的配置,否则会报异常。
- 进行数据库操作事前必须调用close_old_connections函数,避免使用到失效的链接。Django的中间件会自动调用此函数,我们在非Django下操作数据库前要手动调用一次。