13
13
14
14
import os
15
15
16
- from neo4j import GraphDatabase
16
+ from neo4j import Driver , GraphDatabase
17
17
18
18
from testcontainers .core .generic import DbContainer
19
19
from testcontainers .core .waiting_utils import wait_container_is_ready , wait_for_logs
@@ -38,37 +38,33 @@ class Neo4jContainer(DbContainer):
38
38
39
39
# The official image requires a change of password on startup.
40
40
NEO4J_ADMIN_PASSWORD = os .environ .get ("NEO4J_ADMIN_PASSWORD" , "password" )
41
-
42
41
# Default port for the binary Bolt protocol.
43
42
DEFAULT_BOLT_PORT = 7687
44
-
45
43
AUTH_FORMAT = "neo4j/{password}"
46
-
47
44
NEO4J_STARTUP_TIMEOUT_SECONDS = 10
48
-
49
45
NEO4J_USER = "neo4j"
50
46
51
- def __init__ (self , image = "neo4j:latest" , ** kwargs ):
47
+ def __init__ (self , image : str = "neo4j:latest" , ** kwargs ) -> None :
52
48
super (Neo4jContainer , self ).__init__ (image , ** kwargs )
53
49
self .bolt_port = Neo4jContainer .DEFAULT_BOLT_PORT
54
50
self .with_exposed_ports (self .bolt_port )
55
51
self ._driver = None
56
52
57
- def _configure (self ):
53
+ def _configure (self ) -> None :
58
54
self .with_env (
59
55
"NEO4J_AUTH" ,
60
56
Neo4jContainer .AUTH_FORMAT .format (password = Neo4jContainer .NEO4J_ADMIN_PASSWORD )
61
57
)
62
58
63
- def get_connection_url (self ):
59
+ def get_connection_url (self ) -> str :
64
60
return "{dialect}://{host}:{port}" .format (
65
61
dialect = "bolt" ,
66
62
host = self .get_container_host_ip (),
67
63
port = self .get_exposed_port (self .bolt_port ),
68
64
)
69
65
70
66
@wait_container_is_ready ()
71
- def _connect (self ):
67
+ def _connect (self ) -> None :
72
68
# First we wait for Neo4j to say it's listening
73
69
wait_for_logs (
74
70
self ,
@@ -83,7 +79,7 @@ def _connect(self):
83
79
with driver .session () as session :
84
80
session .run ("RETURN 1" ).single ()
85
81
86
- def get_driver (self , ** kwargs ):
82
+ def get_driver (self , ** kwargs ) -> Driver :
87
83
return GraphDatabase .driver (
88
84
self .get_connection_url (),
89
85
auth = (Neo4jContainer .NEO4J_USER , Neo4jContainer .NEO4J_ADMIN_PASSWORD ),
0 commit comments