SQLAlchemy初始化和创建连接

SQLAlchemy初始化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column
from sqlalchemy.types import String, Integer
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('postgresql://test@localhost:5432/test')
DBSession = sessionmaker(engine)
session = DBSession()

BaseModel = declarative_base()

class User(BaseModel):
    __tablename__ = 'user'

    id = Column(String, primary_key=True)
    username = Column(String, index=True)

class Session(BaseModel):
    __tablename__ = 'session'

    id = Column(String, primary_key=True)
    user = Column(String, index=True)
    ip = Column(String)

query = session.query(Session, User.username).join(User, User.id == Session.user)
for i in query:
    print dir(i)

SQLAlchemy创建连接

Engine

engine 的定义包含了三部分的内容:

  1. 具体数据库类型的实现
  2. 连接池
  3. 策略(即engine自己的实现)

所谓的数据库类型即是MYSQL , Postgresql , SQLite 这些不同的数据库.
一般创建 engine 是使用create_engine方法:

1
engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')

对于这个字符串, SQLAlchemy 提供了工具可用于处理它:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# -*- coding: utf-8 -*-

from sqlalchemy import create_engine
from sqlalchemy.engine.url import make_url
from sqlalchemy.engine.url import URL

s = 'postgresql://test@localhost:5432/bbcustom'
url = make_url(s)
s = URL(drivername='postgresql', username='test', password="",host="localhost", port=5432, database="bbcustom")

engine = create_engine(url)
engine = create_engine(s)

print engine.execute('select id from "user"').fetchall()

create_engine 函数有很多的控制参数, 这个后面再详细说.

Engine的策略

create_engine的调用, 实际上会变成strategy.create的调用. 而strategy就是engine的实现细节. strategy可以在create_engine调用时通过strategy参数指定, 目前官方的支持有三种:

  1. plain, 默认的
  2. threadlocal, 连接是线程局部的
  3. mock, 所有的 SQL 语句的执行会使用指定的函数

mock 这个实现, 会把所有的 SQL 语句的执行交给指定的函数来做, 这个函数是由create_engine的 executor 参数指定:

1
2
3
4
5
6
7
def f(sql, *args, **kargs):
    print sql, args, kargs

s = 'postgresql://test@localhost:5432/bbcustom'
engine = create_engine(s, strategy='mock', executor=f)

print engine.execute('select id from "user"')

各数据库实现

各数据库的实现在 SQLAlchemy 中分成了两个部分, 一是数据库的类型, 二是具体数据库中适配的客户端实现. 比如对于 Postgresql 的访问, 可以使用psycopg2, 也可以使用pg8000:

1
2
3
s = 'postgresql+psycopg2://test@localhost:5432/bbcustom'
s = 'postgresql+pg8000://test@localhost:5432/bbcustom'
engine = create_engine(s)

具体的适配工作, 是需要在代码中实现一个 Dialect 类来完成的. 官方的实现在 dialects 目录下.

获取具体的 Dialect 的行为, 则是前面提到的URL对象的get_dialect方法.create_engine时你单传一个字符串,SQLAlchemy自己也会使用make_url得到一个URL的实例).

连接池

SQLAlchemy 支持连接池, 在create_engine时添加相关参数即可使用.

  1. pool_size:连接数
  2. max_overflow:最多多几个连接
  3. pool_recycle:连接重置周期
  4. pool_timeout:连接超时时间

连接池效果:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# -*- coding: utf-8 -*-

from sqlalchemy import create_engine
from sqlalchemy.engine.url import make_url
from sqlalchemy.engine.url import URL

s = 'postgresql://test@localhost:5432/bbcustom'
engine = create_engine(s, pool_size=2, max_overflow=0)


from threading import Thread

def f():
    print engine.execute('select pg_sleep(5)').fetchall()


p = []
for i in range(3):
    p.append(Thread(target=f))

for t in p:
    t.start()

连接池的实现, 在create_engine调用时也可以指定:

1
2
from sqlalchemy.pool import QueuePool
engine = create_engine('sqlite:///file.db', poolclass=QueuePool)

还有:

1
2
3
4
from sqlalchemy.pool import NullPool
engine = create_engine(
         'postgresql+psycopg2://scott:tiger@localhost/test',
          poolclass=NullPool)

或者仅仅是获取连接的方法:

1
2
3
4
5
6
7
8
9
import sqlalchemy.pool as pool
import psycopg2

def getconn():
    c = psycopg2.connect(username='ed', host='127.0.0.1', dbname='test')
    # do things with 'c' to set up
    return c

engine = create_engine('postgresql+psycopg2://', creator=getconn)

连接池可以被单独使用:

1
2
3
4
5
6
7
8
9
10
11
12
import sqlalchemy.pool as pool
import psycopg2

def getconn():
    c = psycopg2.connect(username='ed', host='127.0.0.1', dbname='test')
    return c

mypool = pool.QueuePool(getconn, max_overflow=10, pool_size=5)

conn = mypool.connect()
cursor = conn.cursor()
cursor.execute("select foo")

连接池可以被多个 engine 共享使用:

1
2
e = create_engine('postgresql://', pool=mypool)

展开全文 >>

django-migrate运行python函数

之前代码里面的contact是可以重复的，现在要将其改成unique的，改代码容易，但是还要兼容之前的数据，就要想办法将之前重复的给改掉
方法1: 手动写SQL，在执行到具体的migrations时先执行SQL，但是这个不太统一，另外推代码时容易忘记。

方法2: 框架用的人多就是好，不要重复造轮子，如下: 

1
2
3
4
5
6
7
8
9
 
class Migration(migrations.Migration): 
    def unique_contact(self,*args,**kwargs): 
        # 这个函数在更改contact字段时会先执行到这里 
        #将重复的contact更改掉 
 
    migrations.RunPython(unique_contact), # 这个要在前面    
    migrations.AlterField(model_name='lofter_low',name='contact',field=models.CharField(unique=True, max_length=20),),] 
    

展开全文 >>

2017年终总结

来到公司已经半年了，经历了蛮多的事情，慢慢的从陌生到熟悉到被信任，从部分开发到主力开发，这个过程是非常艰难但又非常开心但又非常有责任的。从与项目经理交流获取业务流程、细节到脑中大致出现业务模型然后到数据库的表结构设计到提供api到测试到文档的编写，每一步都亲力亲为，很多时候又会特地放慢脚步，生怕出现哪个地方没有想到，哪个地方设计的不合理。程序写完并且走完单元测试后，还要写文档给客户端以及Web端，要有分组，哪些给客户端的，哪些给Web端的，哪些接口应该放在一起，怎么将接口排序，哪些接口需不需要验证，需不需要授权，参数怎么传，格式，校验……
最后主要还要在客户端以及Web端看不太明白时还要过去跟他解释、示范，整个过程都是非常的费力以及也是被逐渐熟悉并信任的过程。如果客户端或者web端说某个接口是不是有问题或者测试说这里表现的有些问题，自己还有点心惊胆战以及半信半疑，当发现是他们没有填写对或者没读懂文档或者其它问题导致时，这个过程就慢慢的被人所信任，一旦出现了问题，你的能力就会被怀疑，也许这个责任是比较有压力的，也是感受的比较深的。
目前的我觉得后端担负的责任以及压力还是蛮大的，所有的上层建筑均踩在后端的基础上，想起之前公司总监说后端最终展示出来的可能只是冰山一角。也许这句话有些夸大，但是慢慢的我感觉是对的，所有人都在等你接口，都在等你的实现，虽然我很喜欢后端，但是这个有些倾斜的天平让我觉得有些不适应。
但是我还是很喜欢这个角色，也很喜欢接受更多的挑战，不仅仅是技术方面的，还有在整个团队中的挑战，以前觉得接触更多新的技术可以让自己觉得更具有挑战性，现在有些觉得把事情做好并被认可可能更具备挑战性。我觉得我可以做的更好，也可以担负起更多的责任，慢慢努力，慢慢走！

展开全文 >>

推送

如何定位到某一台设备

根据不同的tag进行定位，例如： 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
{
    "where": {
        "and": [
            {
                "and": [
                    {
                        "tag": "guocheng67"
                    },
                    {
                        "tag": "njy12"
                    },
                    {
                        "tag": "test"
                    }
                ]
            }
        ]
    }
}

上面表示: 发送消息给农技员ID为12、过程为67、环境为test。又由于多台设备可以指定相同的tag，从而单条推送可以推送给多台设备。所以此处的组播意思可能就变了，根据不同的tag组合定位到不同的设备个、群体，实现组播发送，此处tag为app端设置。

关于解决友盟推送每分钟限制

官方文档限制每分钟只能发送5条消息，所以后端维持一个队列A以及计时器，进行分批发送，保证消息发送出去。

此队列实现思想：

  1. 推送请求过来，将此推送消息体保存至redis hash表中，key为uuid,value为消息体
  2. 将此 uuid lpush至队列A，等待任务被执行

此计时器实现思想：

  1. 每隔一段间隔(例如3s)进行轮询存储在队列A中的任务,如果有内容，则定位到当前时间至分钟，例如201712100304,利用redis incr进行操作此key，如果大于5，则表示此分钟已经发送满，则继续轮训至下分钟进行发送
  2. 如果执行成功，则将此uuid rpop至另外一个队列B,实现原子性操作，此处只能表示此任务已经被执行了，但是不能代表执行成功，此处可参考错误

错误

  1. 任务没有被正常执行，或者消息丢失，涉及原因可能多样
  2. 友盟推送返回错误，此处错误可以参考错误

对于错误1，理论应再次推入队列A，但此处不考虑。
对于错误2，除了应用次数限制错误，其余错误可以避免。

任务以及结果存储期限

此处队列与存储均使用redis进行实现，所以理论上只要redis服务不停止，服务状态与结果均可以持久化存储，又由于推送非持久化过程，所以此处暂时满足。

定时发送消息

后端维持一个定时任务，到达时间后，会将消息推送给队列A，等待发送。
此处可以动态增、删任务。

立即发送

将此消息推到队首／尾，等待发送。

后记

转战友盟，此片结束。

展开全文 >>

jenkins使用

嗯嗯，为什么会有这篇文章？？
知道当你写代码的时候前端让你拉代码的痛苦么～～，没有运维我来上，手动多了就更😡了，所以，你懂的。。 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
## 搭建 
  
使用的docker进行部署，但是注意不要部署在8080端口，公司电信有时不能正常访问 
   
## 环境 
  
1. jenkins 
2. gitlab 
  
## 使用 
  
关于jenkins,目前熟悉的流程： 
  
1. gitlab推送消息 
2. jenkins获取消息，拉取代码，todo其他 
   
目前使用步骤为： 
  
+ 安装gitlab webhook等插件，重启jenkins，构建触发器会多出一个可以使用gitlab的.. 
+ 关于gitlab推送，使用secret token 
+ 关于代码部署,使用publish over ssh 
-  配置此jenkins能够访问生产环境的私钥 
-  send files or execute command over ssh那配置参数，此处详情如下。 
-  构建后操作 配置 email通知，如果失败，会发送邮件 
   
构建: 
+ transfters: 
- source files:   **/*  (建议直接填写这个,这个表示这个项目的所有文件) 
- remove prefix: 没用到 
- remote directory: 此对应配置ssh服务器的时候有个remote directory的相对路径 
1. 如果remote directory为/srv,填写project_name, 则最后项目文件存放位置为： /srv/project_name 
2. 注意： 所有文件是拷贝过去的，不是git仓库 
3. 当然，我使用了下面方式： 先手动部署一遍，然后只指定下面的exec command参数 
- exec command: 此处对应生产环境的路径 
  
其他的网上都能找到现成的，此处记录以纪念学习历程。。。

展开全文 >>

protocol-buffters

关于protocol-buffers的语法可以在：
https://developers.google.com/protocol-buffers/docs/proto3
找到，只介绍proto3的语法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
syntax = "proto3";


service Greeter {
    rpc say_hello (hello_request) returns (hello_reply) {
    }
    rpc say_hello_again (hello_request) returns (hello_reply) {
    }
}

message hello_request {
    string name = 1;
    int32 age = 2;
    enum Gender {
        MALE = 0;
        FEMALE = 1;
        UNKNOWN = 2;
    }

    Gender gender = 3;

    repeated string school = 4; // type: list school=["a","b"]
    map<string,string> other = 6; // type: dict other = {"a":"b"}

    /*
        [
            {
                "name": "第一家",
                "address": "address1"
            },
            {
                "name": "第二家",
                "address": "address2"
            }
        ]

    */
    repeated Supermarket supermarket = 7;
}

message Supermarket {
    string name = 1;
    string address = 2;
}

message hello_reply {
    string message = 1;
}

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

探索世界美好的存在。