非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

注意事项:

  1. 引用DjangoORM之前必须要先初始化Django的配置,否则会报异常。
  2. 进行数据库操作事前必须调用close_old_connections函数,避免使用到失效的链接。Django的中间件会自动调用此函数,我们在非Django下操作数据库前要手动调用一次。