利用redis和fastapi实现本地与平台策略进行交互

redis简介:

在pandas一文有详细使用方法(一文教会pandas-CSDN博客)，具体可视化软件有redisstudio等。它是一个由 Salvatore Sanfilippo 写的 key-value 存储系统，是跨平台的非关系型数据库。

Redis 是一个开源的使用 ANSI C 语言编写、遵守 BSD 协议、支持网络、可基于内存、分布式、可选持久性的键值对(Key-Value)存储数据库，并提供多种语言的 API。

Redis 通常被称为数据结构服务器，因为值（value）可以是字符串(String)、哈希(Hash)、列表(list)、集合(sets)和有序集合(sorted sets)等类型。

在命令行中启动redis:

redis-cli.exe

启动不了的时候重启服务：

redis-server

redis事务:

Redis 事务可以一次执行多个命令，并且带有以下三个重要的保证：

批量操作在发送 EXEC 命令前被放入队列缓存。
收到 EXEC 命令后进入事务执行，事务中任意命令执行失败，其余的命令依然被执行。
在事务执行过程，其他客户端提交的命令请求不会插入到事务执行命令序列中。

一个事务从开始到执行会经历以下三个阶段：

开始事务。
命令入队。
执行事务。

如果你想在服务器上运行你随时写的代码，可以将代码写入 Redis 数据库，然后在远程服务器上读取并执行这些代码，通常涉及以下步骤：

序列化代码：将代码转换为一种可以存储在 Redis 中的格式，通常是字符串格式。
写入 Redis：将序列化后的代码存储在 Redis 数据库中。
读取代码：从 Redis 数据库中读取存储的代码。
反序列化代码：将读取到的字符串格式代码转换回可执行的格式。
执行代码：在安全的环境中执行这些代码。

假设你有一段代码：

# 一段函数
def my_function():
    print("This is my function running!")

# 将其convert为字符串：
code_str = inspect.getsource(my_function)

# 一段可执行程序
import inspect
code = '''
a=1
if a>1:
    print("Hello, World!")
else:
    print('error')
'''
    
r.set('code_snippet', code)

# 一个文件
with open('macd_test.py', 'r') as file:
    file_content = file.read()

r.set('python_file_content', file_content)

然后将这个字符串存储在 Redis 中。

import redis

# 连接到 Redis
r = redis.Redis(host='127.0.0.1', port=6379, db=8)

# 将代码存储为 Redis 字符串
r.set('my_function_code', code_str)

在远程服务器上，授权 Redis远程连接权限并读取这个字符串。

# 读取代码
code_str = r.get('my_function_code').decode('utf-8')

反序列化代码，用

exec

或

eval

函数来执行字符串形式的代码：

# 反序列化并执行代码
exec(code_str)

得到运行结果：

这种方式是在你完全信任前提下可以这么做：

使用 exec 或 eval 执行从外部源（如 Redis）读取的代码极其危险，因为这可能导致代码注入和安全漏洞。
如果必须执行从外部存储读取的代码，请确保代码来源是可信的，并且执行环境是安全的。
考虑使用其他方法来存储和执行代码，例如将代码存储在文件系统中，并通过安全的 API 调用来触发代码执行。

在大多数情况下是不会将可执行代码存储在 Redis 中。如果确实需要远程执行代码，通常会有一个专门的服务或 API 来处理代码的执行。

redis 事务的相关命令：

序号

命令及描述

DISCARD
取消事务，放弃执行事务块内的所有命令。

EXEC
执行所有事务块内的命令。

MULTI
标记一个事务块的开始。

UNWATCH
取消 WATCH 命令对所有 key 的监视。

WATCH key [key ...]
监视一个(或多个) key ，如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。

redis连接：

客户端通过密码验证连接到 redis 服务，并检测服务是否在运行：

redis 127.0.0.1:6379> AUTH "password"
OK
redis 127.0.0.1:6379> PING
PONG

redis 连接的基本命令：
序号命令及描述1AUTH password
验证密码是否正确2ECHO message
打印字符串3PING
查看服务是否运行4QUIT
关闭当前连接5SELECT index
切换到指定的数据库

redis服务器：

Redis 服务器命令主要是用于管理 redis 服务。获取 redis 服务器的统计信息：

redis 127.0.0.1:6379> INFO

# Server
redis_version:2.8.13
redis_git_sha1:00000000
redis_git_dirty:0
redis_build_id:c2238b38b1edb0e2
redis_mode:standalone
os:Linux 3.5.0-48-generic x86_64
arch_bits:64
multiplexing_api:epoll
gcc_version:4.7.2
process_id:3856
run_id:0e61abd297771de3fe812a3c21027732ac9f41fe
tcp_port:6379
uptime_in_seconds:11554
uptime_in_days:0
hz:10
lru_clock:16651447
config_file:

# Clients
connected_clients:1
client-longest_output_list:0
client-biggest_input_buf:0
blocked_clients:0

# Memory
used_memory:589016
used_memory_human:575.21K
used_memory_rss:2461696
used_memory_peak:667312
used_memory_peak_human:651.67K
used_memory_lua:33792
mem_fragmentation_ratio:4.18
mem_allocator:jemalloc-3.6.0

# Persistence
loading:0
rdb_changes_since_last_save:3
rdb_bgsave_in_progress:0
rdb_last_save_time:1409158561
rdb_last_bgsave_status:ok
rdb_last_bgsave_time_sec:0
rdb_current_bgsave_time_sec:-1
aof_enabled:0
aof_rewrite_in_progress:0
aof_rewrite_scheduled:0
aof_last_rewrite_time_sec:-1
aof_current_rewrite_time_sec:-1
aof_last_bgrewrite_status:ok
aof_last_write_status:ok

# Stats
total_connections_received:24
total_commands_processed:294
instantaneous_ops_per_sec:0
rejected_connections:0
sync_full:0
sync_partial_ok:0
sync_partial_err:0
expired_keys:0
evicted_keys:0
keyspace_hits:41
keyspace_misses:82
pubsub_channels:0
pubsub_patterns:0
latest_fork_usec:264

# Replication
role:master
connected_slaves:0
master_repl_offset:0
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0

# CPU
used_cpu_sys:10.49
used_cpu_user:4.96
used_cpu_sys_children:0.00
used_cpu_user_children:0.01

# Keyspace
db0:keys=94,expires=1,avg_ttl=41638810
db1:keys=1,expires=0,avg_ttl=0
db3:keys=1,expires=0,avg_ttl=0

序号命令及描述1BGREWRITEAOF
异步执行一个 AOF（AppendOnly File）文件重写操作2BGSAVE
在后台异步保存当前数据库的数据到磁盘3CLIENT KILL [ip:port] [ID client-id]
关闭客户端连接4CLIENT LIST
获取连接到服务器的客户端连接列表5CLIENT GETNAME
获取连接的名称6CLIENT PAUSE timeout
在指定时间内终止运行来自客户端的命令7CLIENT SETNAME connection-name
设置当前连接的名称8CLUSTER SLOTS
获取集群节点的映射数组9COMMAND
获取 Redis 命令详情数组10COMMAND COUNT
获取 Redis 命令总数11COMMAND GETKEYS
获取给定命令的所有键12TIME
返回当前服务器时间13COMMAND INFO command-name [command-name ...]
获取指定 Redis 命令描述的数组14CONFIG GET parameter
获取指定配置参数的值15CONFIG REWRITE
对启动 Redis 服务器时所指定的 redis.conf 配置文件进行改写16CONFIG SET parameter value
修改 redis 配置参数，无需重启17CONFIG RESETSTAT
重置 INFO 命令中的某些统计数据18DBSIZE
返回当前数据库的 key 的数量19DEBUG OBJECT key
获取 key 的调试信息20DEBUG SEGFAULT
让 Redis 服务崩溃21FLUSHALL
删除所有数据库的所有key22FLUSHDB
删除当前数据库的所有key23INFO [section]
获取 Redis 服务器的各种信息和统计数值24LASTSAVE
返回最近一次 Redis 成功将数据保存到磁盘上的时间，以 UNIX 时间戳格式表示25MONITOR
实时打印出 Redis 服务器接收到的命令，调试用26ROLE
返回主从实例所属的角色27SAVE
同步保存数据到硬盘28SHUTDOWN [NOSAVE] [SAVE]
异步保存数据到硬盘，并关闭服务器29SLAVEOF host port
将当前服务器转变为指定服务器的从属服务器(slave server)30SLOWLOG subcommand [argument]
管理 redis 的慢日志31
SYNC

用于复制功能(replication)的内部命令

redis GEO：

Redis GEO 主要用于存储地理位置信息，并对存储的信息进行操作，该功能在 Redis 3.2 版本新增。

Redis GEO 操作方法有：

geoadd：添加地理位置的坐标。
geopos：获取地理位置的坐标。
geodist：计算两个位置之间的距离。
georadius：根据用户给定的经纬度坐标来获取指定范围内的地理位置集合。
georadiusbymember：根据储存在位置集合里面的某个地点获取指定范围内的地理位置集合。
geohash：返回一个或多个位置对象的 geohash 值。

geoadd

geoadd 用于存储指定的地理空间位置，可以将一个或多个经度(longitude)、纬度(latitude)、位置名称(member)添加到指定的 key 中。 key 为 Sicily，Palermo 和 Catania 为位置名称

redis> GEOADD Sicily 13.361389 38.115556 "Palermo" 15.087269 37.502669 "Catania"
(integer) 2
redis> GEODIST Sicily Palermo Catania
"166274.1516"
redis> GEORADIUS Sicily 15 37 100 km
1) "Catania"
redis> GEORADIUS Sicily 15 37 200 km
1) "Palermo"
2) "Catania"
redis>

geopos

geopos 用于从给定的 key 里返回所有指定名称(member)的位置（经度和纬度），不存在的返回 nil。

redis> GEOADD Sicily 13.361389 38.115556 "Palermo" 15.087269 37.502669 "Catania"
(integer) 2
redis> GEOPOS Sicily Palermo Catania NonExisting
1) 1) "13.36138933897018433"
   2) "38.11555639549629859"
2) 1) "15.08726745843887329"
   2) "37.50266842333162032"
3) (nil)
redis>

geodist

geodist 用于返回两个给定位置之间的距离。

redis> GEOADD Sicily 13.361389 38.115556 "Palermo" 15.087269 37.502669 "Catania"
(integer) 2
redis> GEODIST Sicily Palermo Catania
"166274.1516"
redis> GEODIST Sicily Palermo Catania km
"166.2742"
redis> GEODIST Sicily Palermo Catania mi
"103.3182"
redis> GEODIST Sicily Foo Bar
(nil)
redis>

"""
GEODIST key member1 member2 [m|km|ft|mi]
member1 member2 为两个地理位置。

最后一个距离单位参数说明：

m ：米，默认单位。
km ：千米。
mi ：英里。
ft ：英尺。
"""

geohash

Redis GEO 使用 geohash 来保存地理位置的坐标。geohash 用于获取一个或多个位置元素的 geohash 值。

redis> GEOADD Sicily 13.361389 38.115556 "Palermo" 15.087269 37.502669 "Catania"
(integer) 2
redis> GEOHASH Sicily Palermo Catania
1) "sqc8b49rny0"
2) "sqdtr74hyu0"
redis>

redis stream：

Redis Stream 主要用于消息队列（MQ，Message Queue），Redis 本身是有一个 Redis 发布订阅 (pub/sub) 来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis 宕机等，消息就会被丢弃。

简单来说发布订阅 (pub/sub) 可以分发消息，但无法记录历史消息。

而 Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。

Redis Stream 的结构如下所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的 ID 和对应的内容：

每个 Stream 都有唯一的名称，它就是 Redis 的 key，在我们首次使用 xadd 指令追加消息时自动创建。

Consumer Group ：消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer)。
last_delivered_id ：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。
pending_ids ：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符）。

消息队列相关命令：

XADD - 添加消息到末尾
XTRIM - 对流进行修剪，限制长度
XDEL - 删除消息
XLEN - 获取流包含的元素数量，即消息长度
XRANGE - 获取消息列表，会自动过滤已经删除的消息
XREVRANGE - 反向获取消息列表，ID 从大到小
XREAD - 以阻塞或非阻塞方式获取消息列表

消费者组相关命令：

XGROUP CREATE - 创建消费者组
XREADGROUP GROUP - 读取消费者组中的消息
XACK - 将消息标记为"已处理"
XGROUP SETID - 为消费者组设置新的最后递送消息ID
XGROUP DELCONSUMER - 删除消费者
XGROUP DESTROY - 删除消费者组
XPENDING - 显示待处理消息的相关信息
XCLAIM - 转移消息的归属权
**XINFO **- 查看流和消费者组的相关信息；
XINFO GROUPS - 打印消费者组的信息；
XINFO STREAM - 打印流信息

XADD

使用 XADD 向队列添加消息，如果指定的队列不存在，则创建一个队列，XADD 语法格式：

XADD key ID field value [field value ...]

key ：队列名称，如果不存在就创建
ID ：消息 id，我们使用 * 表示由 redis 生成，可以自定义，但是要自己保证递增性。
field value ：记录。

redis> XADD mystream * name Sara surname OConnor
"1601372323627-0"
redis> XADD mystream * field1 value1 field2 value2 field3 value3
"1601372323627-1"
redis> XLEN mystream
(integer) 2
redis> XRANGE mystream - +
1) 1) "1601372323627-0"
   2) 1) "name"
      2) "Sara"
      3) "surname"
      4) "OConnor"
2) 1) "1601372323627-1"
   2) 1) "field1"
      2) "value1"
      3) "field2"
      4) "value2"
      5) "field3"
      6) "value3"
redis>

XTRIM

使用 XTRIM 对流进行修剪，限制长度，语法格式：

XTRIM key MAXLEN [~] count

key ：队列名称
MAXLEN ：长度
count ：数量

127.0.0.1:6379> XADD mystream * field1 A field2 B field3 C field4 D
"1601372434568-0"
127.0.0.1:6379> XTRIM mystream MAXLEN 2
(integer) 0
127.0.0.1:6379> XRANGE mystream - +
1) 1) "1601372434568-0"
   2) 1) "field1"
      2) "A"
      3) "field2"
      4) "B"
      5) "field3"
      6) "C"
      7) "field4"
      8) "D"
127.0.0.1:6379> 

redis>

......

redis分区：

分区是分割数据到多个Redis实例的处理过程，因此每个实例只保存key的一个子集。

分区的优势

通过利用多台计算机内存的和值，允许我们构造更大的数据库。
通过多核和多台计算机，允许我们扩展计算能力；通过多台计算机和网络适配器，允许我们扩展网络带宽。

分区的不足

redis的一些特性在分区方面表现的不是很好：

涉及多个key的操作通常是不被支持的。举例来说，当两个set映射到不同的redis实例上时，你就不能对这两个set执行交集操作。
涉及多个key的redis事务不能使用。
当使用分区时，数据处理较为复杂，比如你需要处理多个rdb/aof文件，并且从多个实例和主机备份持久化文件。
增加或删除容量也比较复杂。redis集群大多数支持在运行时增加、删除节点的透明数据平衡的能力，但是类似于客户端分区、代理等其他系统则不支持这项特性。然而，一种叫做presharding的技术对此是有帮助。

分区类型

Redis 有两种类型分区。假设有4个Redis实例 R0，R1，R2，R3，和类似user:1，user:2这样的表示用户的多个key，对既定的key有多种不同方式来选择这个key存放在哪个实例中。也就是说，有不同的系统来映射某个key到某个Redis服务。

范围分区

最简单的分区方式是按范围分区，就是映射一定范围的对象到特定的Redis实例。

比如，ID从0到10000的用户会保存到实例R0，ID从10001到 20000的用户会保存到R1，以此类推。

这种方式是可行的，并且在实际中使用，不足就是要有一个区间范围到实例的映射表。这个表要被管理，同时还需要各种对象的映射表，通常对Redis来说并非是好的方法。

哈希分区

另外一种分区方法是hash分区。这对任何key都适用，也无需是object_name:这种形式，像下面描述的一样简单：

用一个hash函数将key转换为一个数字，比如使用crc32 hash函数。对key foobar执行crc32(foobar)会输出类似93024922的整数。
对这个整数取模，将其转化为0-3之间的数字，就可以将这个整数映射到4个Redis实例中的一个了。93024922 % 4 = 2，就是说key foobar应该被存到R2实例中。注意：取模操作是取除的余数，通常在多种编程语言中用%操作符实现。

redis管道技术：

Redis是一种基于客户端-服务端模型以及请求/响应协议的TCP服务。这意味着通常情况下一个请求会遵循以下步骤：

客户端向服务端发送一个查询请求，并监听Socket返回，通常是以阻塞模式，等待服务端响应。
服务端处理命令，并将结果返回给客户端。

Redis 管道技术可以在服务端未响应时，客户端可以继续向服务端发送请求，并最终一次性读取所有服务端的响应。

查看 redis 管道，只需要启动 redis 实例并输入以下命令：

$(echo -en "PING\r\n SET runoobkey redis\r\nGET runoobkey\r\nINCR visitor\r\nINCR visitor\r\nINCR visitor\r\n"; sleep 10) | nc localhost 6379

+PONG
+OK
redis
:1
:2
:3

通过使用 PING 命令查看redis服务是否可用，之后我们设置了 runoobkey 的值为 redis，然后获取 runoobkey 的值并使得 visitor 自增 3 次。

在返回的结果中我们可以看到这些命令一次性向 redis 服务提交，并最终一次性读取所有服务端的响应

管道技术最显著的优势是提高了 redis 服务的性能。

一些测试数据，在下面的测试中，我们将使用Redis的Ruby客户端，支持管道技术特性，测试管道技术对速度的提升效果。

require 'rubygems' 
require 'redis'
def bench(descr) 
start = Time.now 
yield 
puts "#{descr} #{Time.now-start} seconds" 
end
def without_pipelining 
r = Redis.new 
10000.times { 
    r.ping 
} 
end
def with_pipelining 
r = Redis.new 
r.pipelined { 
    10000.times { 
        r.ping 
    } 
} 
end
bench("without pipelining") { 
    without_pipelining 
} 
bench("with pipelining") { 
    with_pipelining 
}

"""
从处于局域网中的Mac OS X系统上执行上面这个简单脚本的数据表明，开启了管道操作后，往返延时已经被改善得相当低。
without pipelining 1.185238 seconds 
with pipelining 0.250783 seconds

"""

lpush和rpush

Ipush：将一个或多个值value插入到列表key的表头，并且各个值value按从左到右的顺序依次插入。
rpush：将一个或多个值value插入到列表key的表尾，并且各个值value按从左到右的顺序依次插入。

省流：都是向表中从左到右插入值，lpush是从表头插入，rpush是从表尾插入

使用lpush创建list
lpush list a,b,c
当输入代码后，其真正执行结果相当于按顺序执行三段不同代码
1.lpush key a 
2.lpush key b
3.lpush key c
当输入第1段代码时list为 a     此时a为表头第0个元素
当输入第2段代码时list为 b,a   因为是从左到右插入表头，所以b从左插入到a前面成为表头第0元素
当输入第3段代码时list为 c,b,a 因为c最后执行也是从左到右插入，所以插到b前面成为第0个元素
所以在执行lrange list 0 -1 时结果为
（1）"c"
（2）"b"
（3）"a"

使用rpush创建list
rpush list a,b,c
当输入代码后，其真正执行结果相当于按顺序执行三段不同代码
1.rpush key a 
2.rpush key b
3.rpush key c
当输入第1段代码时list为 a     此时a为表头第0个元素
当输入第2段代码时list为 a,b   因为是从左到右插入表尾，所以b从左插入到a后面成为表尾第1元素
当输入第3段代码时list为 a,b,c 因为c最后执行也是从左到右插入，所以插到b后面成为第2个元素
所以在执行lrange list 0 -1 时结果为
（1）"a"
（2）"b"
（3）"c"

使用lpush创建list
lpush list a,b,c
此时查询结果为
（1）"c"
（2）"b"
（3）"a"
输入rpush list d
此时查询结果为
（1）"c"
（2）"b"
（3）"a"
（4）"d" 因为rpush是从左往右插入表尾，所以直接插到a后面成为最后一个元素

lpop和rpop

lpop key ：移除并获取列表的第一个元素
rpop key ：移除并获取列表最后一个元素

省流：lpop和rpop都是移除并获取被移除的元素，lpop是从表头，rpop是从表尾

现有键名为”list“的列表，包含以下元素[a,b,c,d]
输入lpop list
此时list元素变为[b,c,d],并且返回被移除的元素"a"
继续输入rpop list
此时list元素变为[b,c],并且返回被移除的元素"d"

拓展

当一起使用 lpush 和 lpop 时其作用相当于栈（先进后出）

举例有个list表，现在往里面插入元素
首先输入lpush a
再输入lpush b
再输入lpush c
此时list表为
（1）"c"
（2）"b"
（3）"a"
使用lpop list 后list为
（1）"b"
（2）"a"
此时可以发现最先插入的元素a还在，最后插入的元素c已经被移除，符合栈先进后出，后进先出的规则
在实际应用中可以运用在函数调用栈、撤销操作、浏览器的历史记录等

当一起使用 lpush 和 rpop 时其作用相当于队列（先进先出）

举例有个list表，现在往里面插入元素
首先输入lpush a
再输入lpush b
再输入lpush c
此时list表为
（1）"c"
（2）"b"
（3）"a"
使用rpop list 后list为
（1）"c"
（2）"b"
此时可以发现最先插入的元素a被移除，符合队列先进先出规则

exec function

The exec function in Python is used to execute a string or code object as Python code. It's one of the three ways to execute code in Python, along with eval and compile. The exec function is particularly useful for dynamically executing code at runtime.

Here's how exec works:

exec(source, globals=None, locals=None, /)

source: This is the code to be executed. It can be a string, a code object, or anything that compile would accept.
globals: This is an optional dictionary that supplies the global namespace for the code being executed. If omitted or None, the dictionary of the current global namespace is used.
locals: This is an optional dictionary that supplies the local namespace for the code being executed. If omitted or None, the dictionary of the current local namespace is used.
When you call exec(code_str, {'self': your_class_instance}), here's what happens:

The code_str is the Python code that you want to execute. This code should be a well-formed Python statement or a block of statements.
The {'self': your_class_instance} dictionary is used to provide a local namespace for the code being executed. In this case, you're mapping the keyword self to an instance of YourClass. This is important because when you're executing a method's code, you need to provide the instance (self) on which the method is being called.
Here are some key points about how exec operates:

The code in code_str is executed as if it were the body of a function. This means that any assignments or variable declarations will be local to the namespace provided in locals.
If code_str contains function or class definitions, they will be added to the globals namespace (or the locals namespace if globals is not provided).
If code_str contains expressions that return values (like function calls), exec will return a tuple containing those values. If there are no expressions, exec will return None.
Security note: Be very cautious when using exec, especially if the code to be executed comes from an untrusted source. Untrusted code can execute arbitrary Python code and potentially lead to security vulnerabilities. Always validate and sanitize any user input before using it with exec.

In your code, the exec statement is used to dynamically execute the method calc_indicator on the instance your_class_instance. This is a rare use case for exec, and it's typically done for more advanced scenarios like dynamically modifying or creating methods at runtime.

setattr function

The setattr function is an built-in function in Python that is used to set an attribute value on an object. It takes three arguments:

obj: The object on which to set the attribute.
name: The name of the attribute, as a string.
value: The value to set the attribute to.
Here are some basic usage examples of setattr:

class MyClass:
    def __init__(self, value):
        self.value = value

# Create an instance of MyClass
my_instance = MyClass(10)

# Use setattr to set an attribute
setattr(my_instance, 'new_attribute', 20)  # Equivalent to my_instance.new_attribute = 20

# Print the newly set attribute
print(my_instance.new_attribute)  # Output: 20

# Use setattr to modify an existing attribute
setattr(my_instance, 'value', 30)  # Equivalent to my_instance.value = 30

# Print the modified attribute
print(my_instance.value)  # Output: 30

A key feature of setattr is that it allows U to specify the attribute name as a string, which means U can dynamically decide the attribute name using variables:

attribute_name = 'dynamic_attribute'
attribute_value = 'Dynamic value'

# Dynamically set the attribute
setattr(my_instance, attribute_name, attribute_value)

# Print the dynamically set attribute
print(getattr(my_instance, attribute_name))  # Output: Dynamic value

实现远程服务器与本地代码交互需要用到上面两个点，本地代码不考虑引用平台上类定义属性或方法时可以不用到，但是如果要用到就需要借助 exec和setattr处理工具

本地我在数据库节点8上写了一段计算快均线和慢均线的简单赋值逻辑，将其作为字符串set给key：code_snippet

# example:
code = '''
def calc_indicator1(self) -> None:
    slow_ma = self.kline_generator.producer.sma(self.params_map.slow_period, array=True)
    fast_ma = self.kline_generator.producer.sma(self.params_map.fast_period, array=True)

    self.state_map.slow_ma, self.pre_slow_ma = slow_ma[-1], slow_ma[-2]
    self.state_map.fast_ma, self.pre_fast_ma = fast_ma[-1], fast_ma[-2]
'''
    
redis.set('code_snippet', code)

远程服务器上写了一个加载请求数据函数get_respose：

    def get_respose(self):
        """fastapi"""
        # 发送 GET 请求
        # response = requests.get(self.url)
        # if response.status_code == 200:
        #     # 打印响应内容
        #     self.output("Data received:", response.text)
        # else:
        #     self.output(f"Error: {response.status_code}")

        """redis"""
        redis = StrictRedis(host=self.host, port=self.port, db=self.db)
        try:
            code_str = redis.get('code_snippet').decode('utf-8')
            return code_str
        except:
            self.output('error of calc_indicator!')
            return

然后通过定义一个计算指标calc_indicator函数：

    def calc_indicator(self) -> None:
        """计算指标数据"""
        code_str = self.get_respose()

        exec(code_str)

        match = re.search('def (\\w+)', code_str)

        var_name = str(match.group().split(' ')[1])
        self.output(f"变量名称{var_name}")

        exec('setattr(self, '+'\''+ var_name +'\''+', locals()[\''+var_name+'\'])')

        exec('self.'+var_name+'(self)')

这样的话就可以做到实时在本地修改指标计算逻辑或下单规则或整个策略逻辑去操作平台上运行的策略(运行结果)：

修改过程中可以引入time函数使策略逻辑修改完成再运行。合理运用try-except捕捉Error可以让程序不至于崩溃。

time.sleep(100)

fastap简介：

是一个用于构建 API 的现代、快速（高性能）的 web 框架，专为在 Python 中构建 RESTful API 而设计。

FastAPI 使用 Python 3.8+ 并基于标准的 Python 类型提示。

FastAPI 建立在 Starlette 和 Pydantic 之上，利用类型提示进行数据处理，并自动生成API文档。

FastAPI 于 2018 年 12 月 5 日发布第一版本，以其易用性、速度和稳健性在开发者中间迅速流行起来。FastAPI 支持异步编程，可在生产环境中运行。

FastAPI 特点

高性能：基于Starlette和Pydantic，利用异步（asynchronous）编程，提供出色的性能。
自动文档生成：自动生成交互式API文档，支持Swagger UI和ReDoc，让API的理解和测试更加直观。
类型注解支持：利用Python的类型提示，提供更严格的输入验证和更好的代码提示。
异步支持：支持异步请求处理，使得处理IO密集型任务更加高效。

FastAPI 适用场景

构建API后端：用于构建RESTful API，支持前后端分离的Web应用。
微服务架构：可以作为微服务的后端框架，支持快速开发和部署。
数据处理API：适用于处理数据，接收和返回JSON数据。
实时通信：支持WebSocket，适用于实时通信场景。

fastapi安装：

# 安装 FastAPI 很简单，这里我们使用 pip 命令来安装。
pip install fastapi
# 另外我们还需要一个 ASGI 服务器，生产环境可以使用 Uvicorn 或者 Hypercorn：
pip install "uvicorn[standard]"

main.py

在pycharm里面edit一个demo:main.py

# // main.py
from typing import Union # 导入 Union 类型，用于支持多种数据类型的参数注解

from fastapi import FastAPI

app = FastAPI() # 创建一个 FastAPI 应用的实例，用于定义和管理应用的各个组件，包括路由。

"""
该路由操作使用 @app.get("/") 装饰器，当用户通过 HTTP GET 请求访问根路径时，
将执行 read_root 函数。函数返回一个包含 {"Hello": "World"} 的字典，
这个字典会被 FastAPI 自动转换为 JSON 格式并返回给用户
"""

@app.get("/") # 定义根路径 / 的路由操作
def read_root():
    return {"Hello": "World"}

"""
该路由操作使用 @app.get("/items/{item_id}") 装饰器，当用户通过 HTTP GET 请求访问/items/{item_id} 路径时，将执行 read_item 函数。
函数接受两个参数：
item_id --是路径参数，指定为整数类型。
q -- 是查询参数，指定为字符串类型或空（None）。
函数返回一个字典，包含传入的 item_id 和 q 参数。
q 参数通过 Union[str, None] 表示可以是字符串类型或空，这样就允许在请求中不提供 q 参数。
"""

@app.get("/items/{item_id}") # 定义带路径参数和查询参数的路由操作
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

在命令行中先进入mian.py主目录，运行以下命令以启动应用：

uvicorn main:app --reload

打开浏览器并访问 127.0.0.1:8000就能够看到 FastAPI 自动生成的交互式文档，并在根路径 ("/") 返回的 JSON 响应。

用浏览器访问127.0.0.1:8000/items/5?q=good good study, day day up!将会有以下 JSON 响应**:**

FastAPI 文档

FastAPI 提供了内置的交互式 API 文档，能够轻松了解和测试 API 的各个端点。

这个文档是自动生成的，基于 OpenAPI 规范，支持 Swagger UI 和 ReDoc 两种交互式界面。

在运行 FastAPI 应用时，Uvicorn 同时启动了交互式 API 文档服务。

默认情况下可以通过访问FastAPI - Swagger UI来打开 Swagger UI 风格的文档：

Swagger UI 提供了一个直观的用户界面，用于浏览 API 的各个端点、查看请求和响应的结构，并支持直接在文档中进行 API 请求测试。通过 Swagger UI可以轻松理解每个路由操作的输入参数、输出格式和请求示例。

或者通过FastAPI - ReDoc来打开 ReDoc 风格的文档：清晰简洁的外观能够以可读性强的方式查看 API 的描述、请求和响应。

交互式文档的优势

实时更新：交互式文档会实时更新，反映出应用代码的最新更改。
自动验证：输入参数的类型和格式会得到自动验证，降低了错误的可能性。
便于测试：可以直接在文档中进行 API 请求测试，避免使用其他工具。

在main.py中每修改一次代码服务会自动进行重新加载，因为上面我们用了reload，更改请求后进入FastAPI - Swagger UI会自动生成我们的UI交互界面，交互式 API 文档将会自动更新，并加入新的请求体(FastAPI - ReDoc)：

点击「Try it out」按钮，之后你可以填写参数并直接调用 API：

更新的main函数为：

from typing import Union
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List
app = FastAPI()

# Create a base class
class Item(BaseModel):
    TickData: List[str] = []  # 已赋值，非必需参数
    postion: List[str] = []   # 已赋值，非必需参数
    description: str = None
"""
Define a function that calculates fast and slow averages
The request method is get
The location is in the root directory

import requests

url = "http://127.0.0.1:8000"

response = requests.get(url)

print(response.json())
"""
@app.get("/")
def read_root():
    code_str = '''
    def main_indicator_data(self) -> dict[str, float]:
        self.output(f"Calculation of indicators：{self.state_map.fast_ma, self.state_map.slow_ma}")
        """main chart indicator"""
        return {
            f"MA{self.params_map.fast_period}": self.state_map.fast_ma,
            f"MA{self.params_map.slow_period}": self.state_map.slow_ma
        }
    '''
    return code_str

"""
Define a function that executes a transaction
The request method is get
The execution directory is in the item directory
item_id: random number
"""
@app.get("/items/{item_id}")
def read_item(item_id: int, exec_trade: Union[str, None] = None):
    exec_trade = """
def exec_signal1(self) -> None:
    self.trade_log(f"：Start of execution of trading signals, list of investors{self.filtered_data}")

    for investor_id in self.filtered_data:
        # print(investor_id['InvestorID'])

        position = self.get_position(self.params_map.instrument_id, "1", investor_id['InvestorID'])
        self.output(
            f"investor{investor_id['InvestorID']}instrument{self.params_map.instrument_id}position：{position.net_position}")
        
        # 挂单未成交,账号有待成交/未撤单订单
        # if self.order_id is not None:
        #     try:
        #         self.cancel_order(self.order_id)# not captured
        #     except:
        #         self.output(f"Failure to withdraw an order，order_id:{self.order_id}.")
        #         self.order_id = None

        self.signal_price = 0

        if position.net_position == 0 and not self.end_of_day:
            # 买开，卖开
            if self.sell_signal:
                self.signal_price = -self.last_price

                self.order_id = self.send_order(
                    exchange=self.params_map.exchange,
                    instrument_id=self.params_map.instrument_id,
                    volume=self.params_map.order_volume,
                    investor=investor_id['InvestorID'],
                    price=(price := self.last_price - self.params_map.pay_up),
                    order_direction="sell"
                )

                self.trade_log(f"卖出开仓信号价格: {price}")

            elif self.buy_signal:
                self.signal_price = self.last_price

                self.order_id = self.send_order(
                    exchange=self.params_map.exchange,
                    instrument_id=self.params_map.instrument_id,
                    volume=self.params_map.order_volume,
                    investor=investor_id['InvestorID'],
                    price=(price := self.last_price + self.params_map.pay_up),
                    order_direction="buy"
                )

                self.trade_log(f"买入开仓信号价格: {price}")

        elif position.net_position > 0 and self.sell_signal:  #: 持有多头仓位
            self.signal_price = -self.last_price
            self.output(f"自动处理多头持仓：{position.net_position}")
            self.order_id = self.auto_close_position(
                exchange=self.params_map.exchange,
                instrument_id=self.params_map.instrument_id,
                price=(price := self.last_price - self.params_map.pay_up),
                volume=position.net_position,
                investor=investor_id['InvestorID'],
                order_direction="sell"
            )

            self.trade_log(f"卖出平仓信号价格: {price}")

        elif position.net_position < 0 and self.buy_signal:  #: 持有空头仓位
            self.signal_price = self.last_price
            self.output(f"自动处理空头持仓：{position.net_position}")
            self.order_id = self.auto_close_position(
                exchange=self.params_map.exchange,
                instrument_id=self.params_map.instrument_id,
                price=(price := self.last_price + self.params_map.pay_up),
                volume=abs(position.net_position),
                investor=investor_id['InvestorID'],
                order_direction="buy"
            )
            self.trade_log(f"买入平仓信号价格: {price}")
"""
    return {"item_id": item_id, "exec_trade": exec_trade}

# instantiated
"""
Don't instantiate it again in the function,
otherwise it will be overwritten after writing it in the call and become null, 
call it before the function call
"""

item = Item()

"""
Read K-line data
url = "http://127.0.0.1:8000/0/tick"
"""
@app.get("/items/{item_id}/tick")
def read_item(item_id: int):
    return {"item_id": item_id, "TickData": item.TickData}

# uvicorn main:app --reload

"""
write K-line data
URL: http://127.0.0.1:8000/0/tick

"""
@app.put("/items/{item_id}/tick")
def update_item(item_id: int, it: dict):
    
    item.TickData.append(it)

    return {"item_id": item_id, "TickData": item.TickData}

"""
Read money_postion data
url = "http://127.0.0.1:8000/0/postion"
"""
@app.get("/items/{item_id}/postion")
def read_item(item_id: int):
    
    return {"item_id": item_id, "postion": item.postion}

"""
write money_postion data
URL: http://127.0.0.1:8000/0/postion

"""

@app.put("/items/{item_id}/postion")
def update_item(item_id: int, it: dict):
    
    item.postion.append(it)

    return {"item_id": item_id, "postion": item.postion}

@app.post("/items/")
def create_item(item: Item):
    item.description = '''
    def main_indicator_data(self) -> dict[str, float]:
        self.output(f"计算指标：{self.state_map.fast_ma, self.state_map.slow_ma}")

        return {
            f"MA{self.params_map.fast_period}": self.state_map.fast_ma,
            f"MA{self.params_map.slow_period}": self.state_map.slow_ma
        }
    '''
    return item

更新的main函数内容为：

创建了一个以basemodel类型的类Item，含Tickdata和position列表（已赋默认值，非必需参数），description；以get方式定义了一个读取根目录下计算指标code_str字段的read_root函数，以get方式定义了一个读取根_子目录下执行交易exec_trade字段的read_item函数，其中item_id是路径必需参数（random number (class:int)），将类实例化，以get方式定义了一个读取tick目录下读取K-line data的read_item函数，以put方式定义了一个更新tick目录写入K-line data的update_item函数，以get方式定义了一个读取postion目录下读取money_postion data的read_item函数，以put方式定义了一个更新postion目录写入money_postion data的update_item函数，最后以post方式定义一个带传参数的计算指标的create_item函数。

在策略里面分别定义：

首先初始化：

    def __init__(self) -> None:
        super().__init__()

        self.url = "http://127.0.0.1:8000/items/0"
        self.url_tick = "http://127.0.0.1:8000/items/0/tick"
        self.url_postion = "http://127.0.0.1:8000/items/0/postion"
        self.host = '127.0.0.1'
        self.port = 6379
        self.db = 8
        self.count = 0

在定义交互函数：


    def get_respose(self, call):
        """
        Request Response

        """

        """fastapi"""
        if call == 'exec_trade':
            response = requests.get(self.url)
            if response.status_code == 200:
                # 打印响应内容
                # self.output("fastapi Data received:", response.json()['exec_trade'])
                return response.json()['exec_trade']
            else:
                self.output(f"Error: {response.status_code}")

        elif call == 'calc_ind':
            """redis"""
            redi = StrictRedis(host=self.host, port=self.port, db=self.db)
            try:
                code_str = redi.get('code_snippet').decode('utf-8')
                return code_str
            except:
                self.output('error of calc_indicator!')
                return

    def return_respose(self, call, data):
        """fastapi"""

        if call == 'tick_data_fastapi':
            data['datetime'] = data['datetime'].strftime('%Y-%m-%d %H:%M:%S')
            self.output(f"begin tick_transform fastapi")
            response = requests.put(self.url_tick, json=data)

            """redis"""
        elif call == 'tick_data_redis':
            data['datetime'] = data['datetime'].strftime('%Y-%m-%d %H:%M:%S')
            self.output(f"begin transform redis")
            redi = StrictRedis(host=self.host, port=self.port, db=self.db)
            redi.set(data['datetime'], str(data))#str(pd.DataFrame(data, index=[0]).to_dict(orient='records')[0]))

            """fastapi"""
        elif call == 'money_position':
            self.output(f"begin postion_transform fastapi")
            response = requests.put(self.url_postion, json=data)

            """redis"""
        elif call == 'order_info':
            self.output(f"begin order_transform redis")
            redi = StrictRedis(host=self.host, port=self.port, db=self.db)
            dt = pd.DataFrame(data,index=[0])
            redi.set(dt['order_sys_id'][0], str(data))
            # redi.set(data['datetime'], str(data))#str(pd.DataFrame(data, index=[0]).to_dict(orient='records')[0]))

运行策略后：

数据已经缓存到本地内存数据库可以进行调用和处理转移：

FastAPI 基本路由

在 FastAPI 中，基本路由是定义 API 端点的关键。每个路由都映射到应用程序中的一个函数，用于处理特定的 HTTP 请求，并返回相应的响应。

根路径路由，创建 FastAPI 实例和根路径路由：

@app.get("/")# 使用 @app.get 装饰器创建一个处理根路径的路由
def read_root():
    U code
    pass

路径参数，设置路由的参数：

@app.get("/items/{item_id}")：# 定义了一个路由路径，其中 {item_id} 是路径参数，对应于函数参数 item_id。
def read_item(item_id: int, q: str = None)：# 路由处理函数接受一个整数类型的路径参数 item_id 和一个可选的字符串类型查询参数 q。
    return {"item_id": item_id, "q": q}

FastAPI 请求和响应

在 FastAPI 中，请求（Request）和响应（Response）是与客户端交互的核心。

FastAPI 提供了强大的工具来解析请求数据，并根据需要生成规范的响应。

接下来详细看下 FastAPI 的请求和响应。HTTP 相关内容可以参考：HTTP 请求方法。

传递 GET 请求的参数 http://127.0.0.1:8000/items/5/?q=runoob，返回:

创建一个 /items/ 路由，使用 @app.post 装饰器表示这是一个处理 POST 请求的路由:

from pydantic import BaseModel
from fastapi import FastAPI

app = FastAPI()
class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

使用 Pydantic 模型 Item 定义了一个请求体，包含多个字段，其中一些有默认值，更多 Pydantic 介绍参考：FastAPI Pydantic 模型。

接下来打开 http://127.0.0.1:8000/docs 来进行 POST 测试：填写请求参数就可以得到结果。

响应数据

返回 JSON 数据，路由处理函数返回一个字典，该字典将被 FastAPI 自动转换为 JSON 格式，并作为响应发送给客户端：

传递 GET 请求的参数 http://127.0.0.1:8000/items/5/?q=runoob，返回 JSON 数据；也可以返回一个 Pydantic 模型实例，FastAPI 会自动将其转换为 JSON 格式，并作为响应发送给客户端：

POST 请求，返回的数据格式如下所示：

{
  "name": "runoob",
  "description": " POST 测试",
  "price": 12,
  "tax": 1
}

请求头和 Cookie

使用 Header 和 Cookie 类型注解获取请求头和 Cookie 数据。

from fastapi import Header, Cookie
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
def read_item(user_agent: str = Header(None), session_token: str = Cookie(None)):
    return {"User-Agent": user_agent, "Session-Token": session_token}

以上代码在浏览器访问 **http://127.0.0.1:8000/items/**，返回 JSON 数据：

重定向和状态码

使用 RedirectResponse 实现重定向，将客户端重定向到 /items/ 路由。

from fastapi import Header, Cookie
from fastapi import FastAPI
from fastapi.responses import RedirectResponse

app = FastAPI()

@app.get("/items/")
def read_item(user_agent: str = Header(None), session_token: str = Cookie(None)):
    return {"User-Agent": user_agent, "Session-Token": session_token}

@app.get("/redirect")
def redirect():
    return RedirectResponse(url="/items/")

以上代码在浏览器访问 http://127.0.0.1:8000/redirect/ 会自动跳转到 http://127.0.0.1:8000/items/ 页面。

使用 HTTPException 抛出异常，返回自定义的状态码和详细信息。

以下实例在 item_id 为 42 会返回 404 状态码：

from fastapi import HTTPException

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id == 42:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

自定义响应头

使用 JSONResponse 自定义响应头:

from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    content = {"item_id": item_id}
    headers = {"X-Custom-Header": "custom-header-value"}
    return JSONResponse(content=content, headers=headers)

以上代码在浏览器访问 http://127.0.0.1:8000/items/42/ 页面显示如下

定义 Pydantic 模型

使用 Pydantic 定义一个模型非常简单，只需创建一个继承自 pydantic.BaseModel 的类，并在其中定义字段。字段的类型可以是任何有效的 Python 类型，也可以是 Pydantic 内置的类型。

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

定义了一个名为 Item 的 Pydantic 模型，包含了四个字段：name、description、price 和 tax，name 和 price 是必需的字段，而 description 和 tax 是可选的字段，其中 tax 的默认值为 None。

请求体验证

在 FastAPI 中，可以将 Pydantic 模型用作请求体（Request Body），以自动验证和解析客户端发送的数据。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return item

create_item 路由处理函数接受一个名为 item 的参数，其类型是 Item 模型。FastAPI 将自动验证传入的 JSON 数据是否符合模型的定义，并将其转换为 Item 类型的实例。

查询参数验证

Pydantic 模型还可以用于验证查询参数、路径参数等。

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.get("/items/")
def read_item(item: Item, q: str = Query(..., max_length=10)):
    return {"item": item, "q": q}

read_item 路由处理函数接受一个 Item 模型的实例作为查询参数，以及一个名为 q 的字符串查询参数。通过使用 Query 函数，还可以为查询参数指定更多的验证规则，如最大长度限制。

路径操作依赖项

FastAPI 提供了简单易用，但功能强大的依赖注入系统，这个依赖系统设计的简单易用，可以让开发人员轻松地把组件集成至 FastAPI。

FastAPI 提供了路径操作依赖项（Path Operation Dependencies）的机制，允许你在路由处理函数执行之前或之后运行一些额外的逻辑。

依赖项就是一个函数，且可以使用与路径操作函数相同的参数。

路径操作依赖项提供了一种灵活的方式来组织代码、验证输入、进行身份验证等。

依赖项通常用于两个方面：

预处理（Before）依赖项： 在路由操作函数执行前运行，用于预处理输入数据，验证请求等。
后处理（After）依赖项： 在路由操作函数执行后运行，用于执行一些后处理逻辑，如日志记录、清理等。

依赖注入

依赖注入是将依赖项注入到路由操作函数中的过程。

在 FastAPI 中，通过在路由操作函数参数中声明依赖项来实现依赖注入。

FastAPI 将负责解析依赖项的参数，并确保在执行路由操作函数之前将其传递给函数。

预处理（Before）

在路由操作函数 read_items 中，通过传入 Depends(common_parameters)，用这个依赖项函数，实现在路由执行前预处理输入数据的功能。

from fastapi import Depends, FastAPI, HTTPException
app = FastAPI()

# 定义依赖项：

# 依赖项函数
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 使用依赖项：

# 路由操作函数

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

后处理（After）

在路由操作函数 read_items_after 中，通过传入 Depends(after_request)，用这个后处理依赖项，实现在路由执行后进行额外操作的功能。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

# 依赖项函数
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 路由操作函数
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

# 后处理函数
async def after_request():
    # 这里可以执行一些后处理逻辑，比如记录日志
    pass

# 后处理依赖项
@app.get("/items/", response_model=dict)
async def read_items_after(request: dict = Depends(after_request)):
    return {"message": "Items returned successfully"}

多个依赖项的组合

common_parameters 和 verify_token 是两个不同的依赖项函数，verify_token 依赖于 common_parameters，这种组合依赖项的方式允许在路由执行前先验证一些参数，然后在进行身份验证。

from fastapi import Depends, FastAPI, HTTPException

app = FastAPI()

# 依赖项函数1
def common_parameters(q: str = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

# 依赖项函数2
def verify_token(token: str = Depends(common_parameters)):
    if token is None:
        raise HTTPException(status_code=400, detail="Token required")
    return token

# 路由操作函数
@app.get("/items/")
async def read_items(token: dict = Depends(verify_token)):
    return token

异步依赖项

依赖项函数和后处理函数可以是异步的，允许在它们内部执行异步操作。

get_token 是一个异步的依赖项函数，模拟了一个异步操作。在路由操作函数 read_items 中，用这个异步依赖项函数。

from fastapi import Depends, FastAPI, HTTPException
from typing import Optional
import asyncio

app = FastAPI()

# 异步依赖项函数
async def get_token():
    # 模拟异步操作
    await asyncio.sleep(2)
    return "fake-token"

# 异步路由操作函数
@app.get("/items/")
async def read_items(token: Optional[str] = Depends(get_token)):
    return {"token": token}

标签： fastapi

本文转载自: https://blog.csdn.net/mnwl12_0/article/details/137835063
版权归原作者輕栀所有，如有侵权，请联系我们删除。